#include "util.h"
#include "buffer_trans.h"
#include "bdb.h"
struct bdb* bopen(char *dbname,char *iname,int seq,uint32 flag,struct bparam* param){
    int fd;
	struct stat sbuf;
	char *idxbuf;
	char buf[BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE];
	struct bhd *head;
	struct bdb *db;
	char *dname;
	if(dbname==NULL) return NULL;
	int tmp=seq;
	int w=0;
	while(tmp>0){
	    tmp/=10;
		w++;
	}
	if(w==0) w++;
	w+=strlen(dbname)+strlen(DEF_BDB_SUFFIX)+3;
	dname=malloc(w);
	if(dname==NULL) return NULL;
	snprintf(dname,w,"%s.%d.%s",dbname,seq,DEF_BDB_SUFFIX);
	DEBUG(printf("%s:dname=%s\n",__func__,dname));
	if(flag==0) flag=O_RDWR|O_CREAT;
	if((fd=open(dname,flag,DEF_FILE_MODE))<0){perror("fopen");return NULL;}
	/*FILE *fp=fopen(dname,"a+");
	if(fp==NULL){
	    printf("%s:database file open fail:%s\n",__func__,strerror(errno));
		free(dname);
		return NULL;
	}
	fd=fileno(fp);*/
	if(fstat(fd,&sbuf)<0){perror("fstat");exit(1);}
	if(sbuf.st_size!=0){
	    if(readn(fd,buf,BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE)!=BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE){
		    perror("read");
			free(dname);
			return NULL;
		}
		head=(struct bhd*)buf;
		if((head->magic!=BDB_HEAD_MAGIC)||(Bpgsiz&(Bpgsiz-1))!=0||Bpgsiz<512
		  ||*(buf+BDB_HEAD_SIZE+sizeof(int32))=='\0'){
		    printf("%s:error!bad index file!!magic=%x pgsiz=%d idxname=%s\n",__func__,head->magic,Bpgsiz,buf+BDB_HEAD_SIZE);
			close(fd);
			free(dname);
			return NULL;
		}
		if((db=calloc(1,sizeof(struct bdb)))==NULL){
		    close(fd);
			free(dname);
			return NULL;
		}
	    if((db->head=mmap(NULL,512,PROT_WRITE|PROT_READ,MAP_SHARED,fd,0))==MAP_FAILED){
	        free(db);
			close(fd);
			free(dname);
	        return NULL;
		}
	}else{
	    if(iname==NULL||strlen(iname)>BDB_IDX_NAME_SIZE-1){
		    close(fd);
			free(dname);
			return NULL;
		}
	    db=calloc(1,sizeof(struct bdb));
		ftruncate(fd,512);
		if((db->head=mmap(NULL,512,PROT_WRITE|PROT_READ,MAP_SHARED,fd,0))==MAP_FAILED){
	        free(db);
			close(fd);
			free(dname);
	        return NULL;
	    }
		db->head->magic=BDB_HEAD_MAGIC;
		if(param==NULL){
		    db->Balign=DEF_BDB_ALIGN;
		    db->Bminkey=DEF_BDB_MINKEY;
		    db->Bpgsiz=DEF_BDB_PAGE_SIZE;
		}else{
		    db->Balign=param->align==0?DEF_BDB_ALIGN:param->align;
			db->Bminkey=param->minkey<2?DEF_BDB_MINKEY:param->minkey;
			db->Bpgsiz=param->pgsiz<512?DEF_BDB_PAGE_SIZE:param->pgsiz;
		}
		db->Bmaxpg=(BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE+BDB_FPA_SIZE+db->Bpgsiz-1)/db->Bpgsiz*db->Bpgsiz;
		db->Bflast=0;
		db->Bfstart=-1;
		db->Bnfpg=0;
		db->Bfnum=BDB_FPA_SIZE/sizeof(uint64);
		//db->Bsequence=seq;
		idxbuf=(char*)db->head+BDB_HEAD_SIZE;
		*((int32*)idxbuf)=seq;
		idxbuf+=sizeof(int32);
		*((int32*)idxbuf)=strlen(iname)>BDB_IDX_NAME_SIZE-(sizeof(int32)<<1)?BDB_IDX_NAME_SIZE-(sizeof(int32)<<1):strlen(iname);
		memcpy(idxbuf+sizeof(int32),iname,*((int32*)idxbuf));
		DEBUG(printf("%s:iname=%s\n",__func__,(char*)db->head+BDB_HEAD_SIZE+sizeof(int32)+sizeof(int32)));
	}
	pthread_rwlock_init(&db->rwlock,NULL);
	pthread_mutex_init(&db->iter.lock,NULL);
	db->fpa=(uint64*)((char*)db->head+BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE);
	db->fd=fd;
	//db->fp=fp;
	if(param!=NULL){
	    db->lcc=cachenew(db->fd,param->lcache_nbucket,db->Bpgsiz,param->lcache_maxbytes,param->lcache_swapratio);
		db->pcc=cachenew(db->fd,param->pcache_nbucket,db->Bpgsiz,param->pcache_maxbytes,param->pcache_swapratio);
	}else{
	    db->lcc=cachenew(db->fd,0,db->Bpgsiz,0,0);
	    db->pcc=cachenew(db->fd,0,db->Bpgsiz,0,0);
	}
	db->osiz=(db->Bpgsiz-BLHEADSIZ)/(db->Bminkey<<1)-BRHEADSIZ-sizeof(int16);
	db->dname=dname;
	DEBUG(printf("%s:osiz=%d pgsiz=%d minkey=%d BLHEADSIZ=%d BRHEADSIZ=%d OPHEADSIZ=%d maxpg=%d\n",
	       __func__,db->osiz,db->Bpgsiz,db->Bminkey,BLHEADSIZ,BRHEADSIZ,OPHEADSIZ,db->Bmaxpg));
	return db;
}
struct bdb* bdbopen(char *file,int flag,struct bparam* param){
    struct stat sbuf;
	int fd;
	char *dname,*idxbuf,*iname;
	char buf[BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE];
	int dlen=strlen(file);
	struct bhd *head;
	struct bdb *db;
	//FILE *fp;
	if(file==NULL||dlen==0) return NULL;
    if(flag==0) flag=O_RDWR|O_CREAT;
	/*char *mode;
	if(flag&O_RDONLY){
	    mode="r";
	}else if(flag&O_RDWR){
	    mode="a+";
	}else{
	    mode="a+";
	}*/
	dname=malloc(dlen+1);
	memcpy(dname,file,dlen);
	dname[dlen]='\0';
	iname=dname;
	if((fd=open(dname,flag,DEF_FILE_MODE))<0){perror("fopen");return NULL;}
	/*if((fp=fopen(dname,mode))==NULL){
	    printf("%s:open database file fail\n",__func__);
		free(dname);
		return NULL;
	}
	fd=fileno(fp);*/
	if(fstat(fd,&sbuf)<0){perror("fstat");exit(1);}
	if(sbuf.st_size!=0){
	    if(readn(fd,buf,BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE)!=BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE){
		    perror("read");
			free(dname);
			return NULL;
		}
		head=(struct bhd*)buf;
		if((head->magic!=BDB_HEAD_MAGIC)||(Bpgsiz&(Bpgsiz-1))!=0||Bpgsiz<512
		  ||*(buf+BDB_HEAD_SIZE+sizeof(int32))=='\0'){
		    printf("%s:error!bad index file!!magic=%x pgsiz=%d idxname=%s\n",__func__,head->magic,Bpgsiz,buf+BDB_HEAD_SIZE);
			close(fd);
			free(dname);
			return NULL;
		}
		if((db=calloc(1,sizeof(struct bdb)))==NULL){
		    close(fd);
			free(dname);
			return NULL;
		}
	    if((db->head=mmap(NULL,512,PROT_WRITE|PROT_READ,MAP_SHARED,fd,0))==MAP_FAILED){
	        free(db);
			close(fd);
			free(dname);
	        return NULL;
		}
	}else{
	    if(iname==NULL||strlen(iname)>BDB_IDX_NAME_SIZE-1){
		    close(fd);
			free(dname);
			return NULL;
		}
	    db=calloc(1,sizeof(struct bdb));
		ftruncate(fd,512);
		if((db->head=mmap(NULL,512,PROT_WRITE|PROT_READ,MAP_SHARED,fd,0))==MAP_FAILED){
	        free(db);
			close(fd);
			free(dname);
	        return NULL;
	    }
		db->head->magic=BDB_HEAD_MAGIC;
		if(param==NULL){
		    db->Balign=DEF_BDB_ALIGN;
		    db->Bminkey=DEF_BDB_MINKEY;
		    db->Bpgsiz=DEF_BDB_PAGE_SIZE;
		}else{
		    db->Balign=param->align==0?DEF_BDB_ALIGN:param->align;
			db->Bminkey=param->minkey<2?DEF_BDB_MINKEY:param->minkey;
			db->Bpgsiz=param->pgsiz<512?DEF_BDB_PAGE_SIZE:param->pgsiz;
		}
		db->Bmaxpg=(BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE+BDB_FPA_SIZE+db->Bpgsiz-1)/db->Bpgsiz*db->Bpgsiz;
		db->Bflast=0;
		db->Bfstart=-1;
		db->Bnfpg=0;
		db->Bfnum=BDB_FPA_SIZE/sizeof(uint64);
		idxbuf=(char*)db->head+BDB_HEAD_SIZE;
		*((int32*)idxbuf)=-1;
		idxbuf+=sizeof(int32);
		*((int32*)idxbuf)=strlen(iname)>BDB_IDX_NAME_SIZE-(sizeof(int32)<<1)?BDB_IDX_NAME_SIZE-(sizeof(int32)<<1):strlen(iname);
		memcpy(idxbuf+sizeof(int32),iname,*((int32*)idxbuf));
		DEBUG(printf("%s:iname=%s\n",__func__,(char*)db->head+BDB_HEAD_SIZE+sizeof(int32)+sizeof(int32)));
	}
	pthread_rwlock_init(&db->rwlock,NULL);
	pthread_mutex_init(&db->iter.lock,NULL);
	db->fpa=(uint64*)((char*)db->head+BDB_HEAD_SIZE+BDB_IDX_NAME_SIZE);
	db->fd=fd;
	//db->fp=fp;
	db->lcc=cachenew(db->fd,0,db->Bpgsiz,0,0);
	db->pcc=cachenew(db->fd,0,db->Bpgsiz,0,0);
	db->osiz=(db->Bpgsiz-BLHEADSIZ)/(db->Bminkey<<1)-BRHEADSIZ-sizeof(int16);
	db->dname=dname;
	DEBUG(printf("%s:osiz=%d pgsiz=%d minkey=%d BLHEADSIZ=%d BRHEADSIZ=%d OPHEADSIZ=%d maxpg=%d\n",
	       __func__,db->osiz,db->Bpgsiz,db->Bminkey,BLHEADSIZ,BRHEADSIZ,OPHEADSIZ,db->Bmaxpg));
	return db;
}
struct bstring* colname(struct bdb* db,int *seq){
    char *ptr=(char*)db->head+BDB_HEAD_SIZE;
	if(seq!=NULL) *seq=*((int32*)ptr);
	ptr+=sizeof(int32);
	struct bstring *colname=strcreate(ptr+sizeof(int32),*((int32*)ptr));
	return colname;
}
uint64 bgetfpg(struct bdb *db){
    uint64 offset;
	if(db->Bnfpg>0){
	    if(db->Bfstart>=0){
	        while(db->Bfstart!=db->Bfend){
		        offset=db->fpa[db->Bfstart];
				if(db->Bfend<0) db->Bfend=db->Bfstart;
			    db->Bfstart=(db->Bfstart+1)%db->Bfnum;
				db->Bnfpg--;
				DEBUG(printf("%s:pgno=%d\n",__func__,offset));
			    if(offset<db->Bmaxpg){
				    if(BDB_TRANSACTION_IN(db)) STACK_PUSH(db->newpg,uint64,offset);
					return offset;
				}
		    }
			if(db->Bnfpg==0){
				offset=db->Bmaxpg;
				db->Bmaxpg+=db->Bpgsiz;
				db->Bflast=0;
				db->Bfstart=-1;
				db->Bnfpg=0;
				if(BDB_TRANSACTION_IN(db)) STACK_PUSH(db->newpg,uint64,offset);
				return offset;
			}
			db->Bfend=db->Bfnum;
			if(db->Bfread!=0) offset=db->Bfread;
			else{
			    offset=db->Bflast;
				db->Bflast=0;
			}
			while(offset!=0){
			    while(offset!=0&&db->Bfend>0){
				    struct freepage fp;
					if(offset<db->Bmaxpg)
					    db->fpa[--db->Bfend]=offset;
			        seekread(db->fd,(char*)&fp,sizeof(fp),offset);
				    offset=fp.nextpg;
			    }
				if(db->Bfend==0||db->Bflast==0) break;
				offset=db->Bflast;
				db->Bflast=0;
			}
			#ifdef __DB_DEBUG
		    if(db->Bfend==db->Bfnum){
				db->Bfstart=-1;
				db->Bflast=0;
				printf("%s:error:nfpg=%d\n",__func__,db->Bnfpg);
				exit(1);
			}
			else
			#endif
			{
			    db->Bfread=offset;
				offset=db->fpa[db->Bfend];
				db->Bnfpg--;
				db->Bfstart=db->Bfend+1;
				db->Bfend=0;
			}
		}else{
		    printf("%s:error in get free page:db->Bnfpg=%d\n",__func__,db->Bnfpg);
		    struct freepage fp;
			offset=db->Bflast;
			seekread(db->fd,(char*)&fp,FREEHEAD,offset);
			db->Bflast=fp.nextpg;
		}
	}
	else{
	    offset=db->Bmaxpg;
		db->Bmaxpg+=db->Bpgsiz;
	}
	if(BDB_TRANSACTION_IN(db)) STACK_PUSH(db->newpg,uint64,offset);
	return offset;
}
int bfreepg(struct bdb *db,uint64 pgno){
    struct freepage fp={FREEPAGEMAGIC};
	/*if(db->Bfstart>=0){
	    int i;
		for(i=0;i<db->Bfnum;i++){
		    printf("%s:%d pgno=%d\n",__func__,i,db->fpa[i]);
		}
		printf("%s:fstart=%d fend=%d\n",__func__,db->Bfstart,db->Bfend);
	}*/
	if(BDB_TRANSACTION_IN(db)){
	    STACK_PUSH(db->delpg,uint64,pgno);
		return true;
	}
	DEBUG(printf("%s:pgno=%d maxpg=%d nfpg=%d\n",__func__,pgno,db->Bmaxpg,db->Bnfpg));
	#ifdef __DB_DEBUG
	if(pgno>=db->Bmaxpg){
	    printf("%s:error:pgno=%d db->Bmaxpg=%d\n",__func__,pgno,db->Bmaxpg);
		exit(1);
	}
	#endif
	if(pgno+db->Bpgsiz==db->Bmaxpg){
	    db->Bmaxpg-=db->Bpgsiz;
		return true;
	}
	if(db->Bfstart<0||db->Bnfpg==0){
		db->fpa[0]=pgno;
		db->Bfstart=0;
		db->Bfend=1;
	}else if(db->Bfend>0&&db->Bfend!=db->Bfstart){
		db->fpa[db->Bfend]=pgno;
		db->Bfend=(db->Bfend+1)%db->Bfnum;
		if(db->Bfend==db->Bfstart) db->Bfend=-1;
	}else{
	    fp.nextpg=db->Bflast;
	    db->Bflast=pgno;
		//printf("%s:pgno=%d fp->nextpg=%d\n",__func__,pgno,fp.nextpg);
	    seekwrite(db->fd,(char*)&fp,sizeof(struct freepage),pgno);
	}
	db->Bnfpg++;
	return true;
}
int bgetoverflowpage(struct bdb *db,uint64 pgno,int offset,char *buf,int ln){
    int gotsiz=0;
	char pbuf[BLOCKSIZ];
	char *ptr=buf;
	struct opg *p=NULL;
	int start,cpsiz,rdsiz,presiz=0;
	int osiz=db->Bpgsiz-OPHEADSIZ;
	p=(struct opg*)pbuf;
	if(ln<=0) return 0;
	while(presiz<offset-osiz&&pgno!=0){//locate the page
	    seekread(db->fd,pbuf,OPHEADSIZ,pgno);
		pgno=OPNEXTPG(p);
		presiz+=osiz;
	}
	if(offset+ln<=osiz+presiz)
	    return seekread(db->fd,buf,ln,pgno+offset-presiz+OPHEADSIZ);
	if(db->Bpgsiz>BLOCKSIZ) p=malloc(db->Bpgsiz);
	start=offset-presiz;
	while(pgno!=0&&gotsiz<ln){
	    rdsiz=ln-gotsiz>db->Bpgsiz-OPHEADSIZ?db->Bpgsiz:ln-gotsiz+OPHEADSIZ;
		if(seekread(db->fd,(char*)p,rdsiz,pgno)!=rdsiz){
		    if((char*)p!=buf) free(p);
			return gotsiz;
		}
		cpsiz=osiz-start>ln-gotsiz?ln-gotsiz:osiz-start;
		memcpy(ptr,pbuf+OPHEADSIZ+start,cpsiz);
		ptr+=cpsiz;
		gotsiz+=cpsiz;
		start=0;
		pgno=OPNEXTPG(p);
	}
	if((char*)p!=pbuf) free(p);
	return gotsiz;
}
uint64 bwriteoverflowpage(struct bdb *db,char *key,int ksiz,char *val,int vsiz){   
	char *ptr=NULL,*buff;
	int siz=0,cpsiz,osiz=db->Bpgsiz-OPHEADSIZ;
	int tot=ksiz+vsiz;
	uint64 pgno,nextpgno,retpgno;
	char buf[BLOCKSIZ];
	if(tot<=0) return true;
	nextpgno=bgetfpg(db);
	retpgno=nextpgno;
	if(db->Bpgsiz>BLOCKSIZ) buff=malloc(db->Bpgsiz);
	else buff=buf;
	struct opg *p=(struct opg*)buff;
	while(siz<tot){
	    ptr=buff+OPHEADSIZ;
		p->siz=0;
		pgno=nextpgno;
		if(tot-siz<=osiz){
		    cpsiz=ksiz-siz;
		    if(cpsiz>0){
 			    memcpy(ptr,key+siz,cpsiz);
			    ptr+=cpsiz;
				siz+=cpsiz;
				p->siz+=cpsiz;
			}
			if(siz<tot){
			    cpsiz=tot-siz;
			    memcpy(ptr,val+(siz-ksiz),cpsiz);
				ptr+=cpsiz;
				p->siz+=cpsiz;
				siz=tot;
			}
			p->nextpg=0;
		}
		else{
	        if(siz<ksiz){
		        cpsiz=ksiz-siz>osiz?osiz:ksiz-siz;
			    memcpy(ptr,key+siz,cpsiz);
			    siz+=cpsiz;
			    ptr+=cpsiz;
				p->siz+=cpsiz;
		    }
		    if(ptr-buff<db->Bpgsiz&&siz<tot){
		        cpsiz=tot-siz>db->Bpgsiz-(ptr-buff)?db->Bpgsiz-(ptr-buff):tot-siz;
		        memcpy(ptr,val+siz-ksiz,cpsiz);
		        siz+=cpsiz;
		        ptr+=cpsiz;
				p->siz+=cpsiz;
		    }
			nextpgno=p->nextpg=bgetfpg(db);
		}
		seekwrite(db->fd,buff,ptr-buff,pgno);
		DEBUG(printf("%s:write overflowpage=%d bytes=%d\n",__func__,pgno,p->siz));
	}
	if(buff!=buf)free(buff);
	return retpgno;
}
int bdeloverflowpage(struct bdb *db,uint64 pgno,int offset,int siz){
    if(pgno==0) return true;
	char buf[OPHEADSIZ];
	struct opg *p=(struct opg*)buf;
	char ref=0;
	uint64 npgno;
	if(offset!=0){
	    while(pgno!=0&&offset>db->Bpgsiz-OPHEADSIZ){
	        seekread(db->fd,(char*)p,OPHEADSIZ,pgno);
		    pgno=OPNEXTPG(p);
			ref=OPREFERRED(p)|ref;
		    offset-=p->siz;
	    }
	    if(pgno==0){
		    printf("%s:nextpg=0\n",__func__);
		    return false;
		}
	    seekread(db->fd,(char*)p,OPHEADSIZ,pgno);
	    npgno=OPNEXTPG(p);
		p->nextpg=0;
	    if(ref) OPREFER(p);
	    seekwrite(db->fd,(char*)p,OPHEADSIZ,pgno);
		pgno=npgno;
	}else{
	    seekread(db->fd,(char*)p,OPHEADSIZ,pgno);
		npgno=OPNEXTPG(p);
	    bfreepg(db,pgno);
		pgno=npgno;
	}
	while(pgno!=0&&siz>0){
		seekread(db->fd,(char*)p,OPHEADSIZ,pgno);
		siz-=p->siz;
		bfreepg(db,pgno);
		pgno=OPNEXTPG(p);
	}
	return true;
}
#define REC(rec,ksize,vsize,fp)\
    do{\
	    (rec)->ksiz=(ksize);\
		(rec)->sop=(vsize);\
		(rec)->fap=(fp);\
	}while(false)
struct brec* breccreate(struct bdb *db,char *key,int ksiz,char *val,int vsiz,char *rbuf,int *rsiz){
    struct brec *rec=(struct brec*)rbuf;
	if(ksiz+vsiz<=db->osiz){
	    int rlen=BRKVSIZ(ksiz,vsiz);
		if(rlen>*rsiz)
		    rec=malloc(rlen);
		*rsiz=rlen;
		char *d=(char*)rec+BRHEADSIZ;
		memcpy(d,key,ksiz);
		memcpy(d+ksiz,val,vsiz);
		REC(rec,ksiz,vsiz,0);
		DEBUG(printf("%s:rlen=%d\n",__func__,rlen));
	}else if(ksiz+sizeof(uint64)<=db->osiz){
	    int rlen=BRKSIZ(ksiz);
		if(rlen>*rsiz) rec=malloc(rlen);
		*rsiz=rlen;
		char *d=(char*)rec+BRHEADSIZ;
		DEBUG(printf("%s:VOFLOW:risz=%d\n",__func__,*rsiz));
		uint64 offset=bwriteoverflowpage(db,NULL,0,val,vsiz);
		*((uint64*)d)=offset;
		memcpy(d+sizeof(uint64),key,ksiz);
		REC(rec,ksiz,vsiz,B_VOFLOW);
	}else{
	    int rlen=db->osiz;
		if(rlen>*rsiz) rec=malloc(rlen);
		*rsiz=rlen;
		char *d=(char*)rec+BRHEADSIZ;
		int pksiz=rlen-sizeof(uint64)-BRHEADSIZ;
		uint64 offset=bwriteoverflowpage(db,key+pksiz,ksiz-pksiz,val,vsiz);
		DEBUG(printf("%s:PKOFLOW:rsiz=%d pksiz=%d offset=%d\n",__func__,*rsiz,pksiz,offset));
		*((uint64*)d)=offset;
		memcpy(d+sizeof(int64),key,pksiz);
		REC(rec,ksiz,vsiz,B_PKOFLOW);
	}
	return rec;
}
int brecsiz(struct bdb *db,struct brec *rec){
    uint16 flag=rec->fap&B_FMASK;
	if(flag==0)return rec->ksiz+rec->sop+BRHEADSIZ;
	else if(flag==B_VOFLOW) return rec->ksiz+sizeof(uint64)+BRHEADSIZ;
	else if(flag&B_PKOFLOW) return db->osiz;
	else if(flag==B_VDUP) return rec->ksiz+BRHEADSIZ;
	printf("%s:error:rec->fap=%x\n",__func__,rec->fap);
	exit(1);
}
#define BIDX(index,ksize,fp)\
    do{\
	    (index)->ksiz=(ksize);\
		(index)->fap=(fp);\
	}while(false)
struct bidx *bidxcreate(struct bdb *db,struct brec *rec,char *ibuf,int *isiz){
    struct bidx *index=(struct bidx*)ibuf;
	int flag=rec->fap&B_FMASK&~B_VDUP;
	int ilen;
	char *d,*s;
	if(flag==0){
	    ilen=rec->ksiz+BIHEADSIZ;
		if(ilen>*isiz) index=malloc(ilen);
		*isiz=ilen;
		d=(char*)index+BIHEADSIZ;
		s=(char*)rec+BRHEADSIZ;
		memcpy(d,s,rec->ksiz);
		BIDX(index,rec->ksiz,0);
	}else if(flag==B_VOFLOW){
	    ilen=rec->ksiz+BIHEADSIZ;
		if(ilen>*isiz) index=malloc(ilen);
		*isiz=ilen;
		d=(char*)index+BIHEADSIZ;
		s=(char*)rec+BRHEADSIZ+sizeof(uint64);
		memcpy(d,s,rec->ksiz);
		BIDX(index,rec->ksiz,0);
	}else if(flag==B_PKOFLOW){
	    ilen=db->osiz;
		if(ilen>*isiz) index=malloc(ilen);
		*isiz=ilen;
		int cpsiz=db->osiz-BRHEADSIZ;
		d=(char*)index+BIHEADSIZ;
		s=(char*)rec+BRHEADSIZ;
		memcpy(d,s,cpsiz);
		BIDX(index,rec->ksiz,B_PKOFLOW);
		char buf[OPHEADSIZ];
		seekread(db->fd,buf,OPHEADSIZ,*((uint64*)((char*)rec+BRHEADSIZ)));
		OPREFER((struct opg*)buf);
		seekwrite(db->fd,buf,OPHEADSIZ,*((uint64*)((char*)rec+BRHEADSIZ)));
	}else{
	    printf("%s:error:rec->fap=%x\n",__func__,rec->fap);
		exit(1);
	}
	return index;
}
#define BC(bc,off,ioff,pg)\
    do{\
	    (bc)->offset=(off);\
		(bc)->idxoff=(ioff);\
		(bc)->p=(pg);\
	}while(false)
int breccmp(struct bdb *db,char *key,int ksiz,struct brec *rec){
    int flag=rec->fap&B_FMASK&~B_VDUP;
	int cmp;
	char *nkey;
	int cmpsiz;
	if(flag==0){
	    nkey=(char*)rec+BRHEADSIZ;
		cmpsiz=rec->ksiz>ksiz?ksiz:rec->ksiz;
	    cmp=strncmp(key,nkey,cmpsiz);
		if(cmp==0) cmp=ksiz-rec->ksiz;
		DEBUG(printf("%s:rec key=%s cmpsiz=%d cmp=%d ksiz=%d rec->siz=%d\n",__func__,nkey,cmpsiz,cmp,ksiz,rec->ksiz));
	}else if(flag==B_VOFLOW){
	    nkey=(char*)rec+BRHEADSIZ+sizeof(uint64);
		cmpsiz=rec->ksiz>ksiz?ksiz:rec->ksiz;
		cmp=strncmp(key,nkey,cmpsiz);
		if(cmp==0) cmp=ksiz-rec->ksiz;
	}else if(flag==B_PKOFLOW){
	    nkey=(char*)rec+BRHEADSIZ+sizeof(uint64);
		int tsiz=db->osiz-BRHEADSIZ-sizeof(uint64);
		cmpsiz=rec->ksiz>tsiz?tsiz:rec->ksiz;
		cmp=strncmp(key,nkey,cmpsiz);
		if(cmp==0){
		    if(ksiz>cmpsiz){
			    char buf[BLOCKSIZ],*okey;
				int nsiz=rec->ksiz-tsiz;
				if(BLOCKSIZ<nsiz) okey=malloc(nsiz);
				else okey=buf;
			    uint64 off=*((uint64*)((char*)rec+BRHEADSIZ));
			    bgetoverflowpage(db,off,0,okey,nsiz);
				int csiz=cmpsiz;
				cmpsiz=(rec->ksiz>ksiz?ksiz:rec->ksiz)-cmpsiz;
				cmp=strncmp(key+csiz,okey,cmpsiz);
				if(cmpsiz==0) cmpsiz=ksiz-rec->ksiz;
				if(okey!=buf) free(okey);
			}else cmp=-1;
		}
	}else{
	    printf("%s:error:fap=%x\n",__func__,rec->fap);
		exit(1);
	}
	return cmp;
}
int bidxcmp(struct bdb *db,char *key,int ksiz,struct bidx *idx){
    int flag=idx->fap&B_FMASK;
	int cmp,cmpsiz;
	char *nkey;
	if(flag==0){
	    nkey=(char*)idx+BIHEADSIZ;
		cmpsiz=idx->ksiz>ksiz?ksiz:idx->ksiz;
		cmp=strncmp(key,nkey,cmpsiz);
		if(cmp==0) cmp=ksiz-idx->ksiz;
	}else if(flag==B_PKOFLOW){
	    nkey=(char*)idx+BIHEADSIZ+sizeof(uint64);
		int tsiz=db->osiz-BIHEADSIZ-sizeof(uint64);
		cmpsiz=tsiz>ksiz?ksiz:tsiz;
		cmp=strncmp(key,nkey,cmpsiz);
		if(cmp==0){
		    if(ksiz>cmpsiz){
			    char buf[BLOCKSIZ],*okey;
				int nsiz=idx->ksiz-tsiz;
				if(nsiz>BLOCKSIZ) okey=malloc(nsiz);
				else okey=buf;
				uint64 off=*((uint64*)((char*)idx+BIHEADSIZ));
				bgetoverflowpage(db,off,0,okey,nsiz);
				int csiz=cmpsiz;
				cmpsiz=(ksiz>idx->ksiz?idx->ksiz:ksiz)-csiz;
				cmp=strncmp(key+csiz,okey,cmpsiz);
				if(cmp==0) cmp=ksiz-idx->ksiz;
			}else cmp=-1;
		}
	}else{
	    printf("%s:error:fap=%x\n",__func__,idx->fap);
		exit(1);
	}
	return cmp;
}
int pgsearch(struct bdb *db,struct bpg *pg,char *key,int ksiz){
    int inum=(pg->lower-BPHEADSIZ)/sizeof(int16);
    DEBUG(printf("%s:inner page inum=%d\n",__func__,inum));
	int left=0,mid,cmp,right=inum-1;
	struct bidx *idx;
	while(left<=right){
		mid=(left+right)>>1;
		idx=(struct bidx*)(pg->index[mid]+(char*)pg);
		if((cmp=bidxcmp(db,key,ksiz,idx))>0){
			left=mid+1;
		}else if(cmp<0){
			right=mid-1;
	    }else{
			left=mid+1;
			break;
		}
	}
	return left;
}
int lfsearch(struct bdb *db,struct blf *l,char *key,int ksiz,int *hit){
	int rnum=(l->lower-BLHEADSIZ)/sizeof(int16);
	int mid,left=0,right=rnum-1,cmp;
	struct brec *rec;
	DEBUG(printf("%s:rnum=%d\n",__func__,rnum));
	while(left<=right){
	    mid=(left+right)>>1;
		rec=(struct brec*)((char*)l+l->index[mid]);
		if((cmp=breccmp(db,key,ksiz,rec))>0){
		    left=mid+1;
		}else if(cmp<0){
		    right=mid-1;
		}else{
		    *hit=1;
			return mid;
		}
	}
	*hit=0;
	return left;
}
int hsearch(struct bdb *db,char *key,int ksiz,int *hit,struct bc *lc){
	if(db->lastpg==0) return false;
    struct blf *l=(struct blf*)getpage(db->lcc,OFFSET(db->lastpg),F_READ);
	int rnum=(l->lower-BLHEADSIZ)/sizeof(int16);
	struct brec *rec=(struct brec*)((char*)l+l->index[rnum-1]);
	int cmp;
	DEBUG(printf("%s:lastpg=%d rnum=%d\n",__func__,db->lastpg,rnum));
	if((cmp=breccmp(db,key,ksiz,rec))>0){
	    *hit=0;
		if(l->next==0){
		    BC(lc,db->lastpg,rnum,l);
			if(db->dir>=0) db->dir++;
			else if(db->dir<0) db->dir=0;
			return true;
		}else{
		    putpage(db->lcc,(char*)l,0);
			db->dir=0;
			return false;
		}
	}else if(cmp==0){
	    *hit=1;
		BC(lc,db->lastpg,rnum-1,l);
		db->dir=0;
		return true;
	}
	rec=(struct brec*)((char*)l+l->index[0]);
	if((cmp=breccmp(db,key,ksiz,rec))<0){
	    *hit=0;
		if(l->pre==0){
		    BC(lc,db->lastpg,0,l);
			if(db->dir<=0) db->dir--;
			else if(db->dir>0) db->dir=0;
			return true;
		}
		else{
		   putpage(db->lcc,(char*)l,0);
		   db->dir=0;
		   return false;
		}
	}else if(cmp==0){
	    *hit=1;
		BC(lc,db->lastpg,0,l);
		db->dir=0;
		return true;
	}
	int ipos=lfsearch(db,l,key,ksiz,hit);
	db->dir=0;
	BC(lc,db->lastpg,ipos,l);
	return true;
}
int leftsearch(struct bdb *db,char *key,int ksiz,int *hit,struct bc *lc){
    struct blf *lf;
	int rnum,cmp;
	struct brec *rec;
	db->dir=0;
	lf=(struct blf*)getpage(db->lcc,OFFSET(db->Blmost),F_READ);
	rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
	DEBUG(printf("%s:lmost=%d rmost=%d lmostrnum=%d\n",__func__,db->Blmost,db->Brmost,rnum));
	rec=(struct brec*)(lf->index[rnum-1]+(char*)lf);
	if((cmp=breccmp(db,key,ksiz,rec))>0){
	    putpage(db->lcc,(char*)lf,0);
		return false;
	}else{
		int ipos=lfsearch(db,lf,key,ksiz,hit);
		BC(lc,db->Blmost,ipos,lf);
		db->lastpg=db->Blmost;
		return true;
	}
}
int rightsearch(struct bdb *db,char *key,int ksiz,int *hit,struct bc *lc){
    struct blf *lf;
	int rnum,cmp,ipos;
	struct brec *rec;
	db->dir=0;
	lf=(struct blf*)getpage(db->lcc,OFFSET(db->Brmost),F_READ);
	rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
	rec=(struct brec*)(lf->index[0]+(char*)lf);
	DEBUG(printf("%s:rmost=%d rnum=%d\n",__func__,db->Brmost,rnum));
	if((cmp=breccmp(db,key,ksiz,rec))<0){
	    putpage(db->lcc,(char*)lf,0);
		return false;
	}else{
		ipos=lfsearch(db,lf,key,ksiz,hit);
		BC(lc,db->Brmost,ipos,lf);
		db->lastpg=db->Brmost;
		return true;
	}
}
#define BSTACKPUSH(db,off,ioff)\
    do{\
	    if((db)->snum==(db)->ssiz){\
		    (db)->ssiz<<=1;\
		    (db)->stack=realloc((db)->stack,(db)->ssiz);\
		}\
		(db)->stack[(db)->snum].offset=(off);\
		(db)->stack[(db)->snum].idxoff=(ioff);\
		(db)->snum++;\
	}while(false)
struct blf* rsearch(struct bdb *db,char *key,int ksiz,int *hit){
    struct bpg *pg;
	struct blf *lf;
	struct bidx *idx;
	int ipos;
	uint64 off=db->Broot;
	db->dir=0;
	if(off==0) return NULL;
	if(db->ssiz==0||db->stack==NULL){
	    db->stack=malloc(sizeof(struct bpstack)*DEF_BDB_LEVEL);
		db->ssiz=DEF_BDB_LEVEL;
	}
	db->snum=0;
	while(!ISBLEAF(off)){
	    pg=(struct bpg*)getpage(db->pcc,off,F_READ);
		ipos=pgsearch(db,pg,key,ksiz);
		BSTACKPUSH(db,off,ipos);
		if(ipos<1) off=pg->left;
		else{
		    idx=(struct bidx*)(pg->index[ipos-1]+(char*)pg);
			off=idx->offset;	
		}
		putpage(db->pcc,(char*)pg,0);
	}
	lf=(struct blf*)getpage(db->lcc,OFFSET(off),F_READ);
	ipos=lfsearch(db,lf,key,ksiz,hit);
	db->lastpg=off;
	BSTACKPUSH(db,off,ipos);
	#ifdef __DB_DEBUG
	int i;
	for(i=0;i<db->snum;i++){
	    printf("%s:stack %d :pgno=%d idxoff=%d\n",__func__,i,db->stack[i].offset,db->stack[i].idxoff);
	}
	#endif
	return lf;
}
int getstack(struct bdb *db,char *key,int ksiz){
    struct bidx *idx;
	int ipos;
	struct bpg* pg;
	uint64 off=db->Broot;
    if(off==0) return 0;
	if(db->ssiz==0||db->stack==NULL){
	    db->stack=malloc(sizeof(struct bpstack)*DEF_BDB_LEVEL);
		db->ssiz=DEF_BDB_LEVEL;
	}
	db->snum=0;
	while(!ISBLEAF(off)){
	    DEBUG(printf("%s:inner pgno=%d\n",__func__,off));
	    pg=(struct bpg*)getpage(db->pcc,off,F_READ);
		ipos=pgsearch(db,pg,key,ksiz);
		BSTACKPUSH(db,off,ipos);
		if(ipos<1) off=pg->left;
		else{
		    idx=(struct bidx*)(pg->index[ipos-1]+(char*)pg);
			off=idx->offset;	
		}
		putpage(db->pcc,(char*)pg,0);
	}
	return true;
}
int bput(struct bdb *db,char *key,int ksiz,char *val,int vsiz,int flag){
    uint64 off;
	struct brec *rec=NULL;
	int rsiz=BLOCKSIZ;
	char buf[BLOCKSIZ];
	char pbuf[BLOCKSIZ];
	struct blf *l=(struct blf*)pbuf;
	printf("%s:key=%s ksiz=%d\n",__func__,key,ksiz);
	pthread_rwlock_wrlock(&db->rwlock);
	if(db->Broot==0){
	    int rsiz=BLOCKSIZ;
	    off=bgetfpg(db);
		db->Broot=BLEAF(off);
		rec=breccreate(db,key,ksiz,val,vsiz,buf,&rsiz);
		DEBUG(printf("%s:root page create %d. rsiz=%d rec->key=%s\n",__func__,off,rsiz,(char*)rec+BRHEADSIZ));
		if(db->Bpgsiz>BLOCKSIZ) l=malloc(db->Bpgsiz);
		l->pre=l->next=0;
		l->lower=BLHEADSIZ+sizeof(int16);
		l->upper=db->Bpgsiz-rsiz;
		l->index[0]=l->upper;
		db->lastpg=db->Brmost=db->Blmost=db->Broot;
		memcpy((char*)l+l->upper,rec,rsiz);
		addpage(db->lcc,off,(char*)l,F_DIRTY);
		if((char*)rec!=buf) free(rec);
		if((char*)l!=pbuf) free(l);
		pthread_rwlock_unlock(&db->rwlock);
		return true;
	}
	int hit=0;
	struct bc cursor;
	if(hsearch(db,key,ksiz,&hit,&cursor)||(OFFSET(db->lastpg)!=db->Blmost&&leftsearch(db,key,ksiz,&hit,&cursor))||(OFFSET(db->lastpg)!=db->Brmost&&rightsearch(db,key,ksiz,&hit,&cursor))){
	    struct blf *lf=cursor.p;
		int ipos=cursor.idxoff;
		if(hit){
		    DEBUG(printf("%s:same key hitted.ipos=%d\n",__func__,ipos));
			adddupval(db,lf,ipos,val,vsiz);
			putpage(db->lcc,(char*)lf,F_DIRTY);
			pthread_rwlock_unlock(&db->rwlock);
			return true;
		}else{
			rec=breccreate(db,key,ksiz,val,vsiz,buf,&rsiz);
			DEBUG(printf("%s:history hit or left right hit.remsiz=%d rsiz=%d ipos=%d\n",__func__,lf->upper-lf->lower,rsiz,ipos);)
			if(rsiz+sizeof(int16)<=lf->upper-lf->lower){
			    int inum=(lf->lower-BLHEADSIZ)/sizeof(int16);
			    lf->upper-=rsiz;
				lf->lower+=sizeof(int16);
				memmove(lf->index+ipos+1,lf->index+ipos,sizeof(int16)*(inum-ipos));
				lf->index[ipos]=lf->upper;
				memcpy((char*)lf+lf->upper,rec,rsiz);
				putpage(db->lcc,(char*)lf,F_DIRTY);
				if((char*)rec!=buf) free(rec);
				pthread_mutex_lock(&db->iter.lock);
				if((db->iter.flag&BI_INUSE)&&db->iter.offset==cursor.offset
				    &&db->iter.idxoff>=ipos)
					 db->iter.idxoff++;
				pthread_mutex_unlock(&db->iter.lock);
				pthread_rwlock_unlock(&db->rwlock);
				return true;
			/*}else if(db->lastpg==cursor.offset&&db->snum!=0){//we can reuse the stack..but this case is rare
				printf("%s:db->snum=%d\n",__func__,db->snum);
				bputrec(db,lf,rec,rsiz);
				if((char*)rec!=buf) free(rec);
				return true;*/
			}else{
			    getstack(db,key,ksiz);
				BSTACKPUSH(db,cursor.offset,ipos);
				//printf("%s:get stack:db->snum=%d remsiz=%d\n",__func__,db->snum,lf->upper-lf->lower);
				bputrec(db,lf,rec,rsiz);
				pthread_rwlock_unlock(&db->rwlock);
				return true;
			}
		}
	}
	l=rsearch(db,key,ksiz,&hit);
	if(hit==0){
	    if(rec==NULL){
		    rsiz=BLOCKSIZ;
		    rec=breccreate(db,key,ksiz,val,vsiz,buf,&rsiz);
		}
	    bputrec(db,l,rec,rsiz);
	}else{
	    int ipos=db->stack[db->snum-1].idxoff;
	    adddupval(db,l,ipos,val,vsiz);
		putpage(db->lcc,(char*)l,F_DIRTY);
	}
	pthread_rwlock_unlock(&db->rwlock);
	return true;
}
int adddupval(struct bdb *db,struct blf *lf,int16 ipos,char *val,int vsiz){
    struct brec *rec=(struct brec*)((char*)lf+lf->index[ipos]);
	int flag=rec->fap&B_FMASK;
	uint64 vdoff,off;
	if(flag&B_VDUP){
		vdoff=rec->sop;
		rec->sop=putvdup(db,vdoff,val,vsiz);
	}else if(flag==0){
		char *oval=(char*)rec+BRHEADSIZ+rec->ksiz;
		DEBUG(printf("%s:oval=%s\n",__func__,oval));
		vdoff=putvdup(db,0,oval,rec->sop);
		vdoff=putvdup(db,vdoff,val,vsiz);
		dellfval(db,lf,ipos);
		rec=(struct brec*)((char*)lf+lf->index[ipos]);
		rec->sop=vdoff;
		rec->fap=B_VDUP;
	}else if(flag==B_VOFLOW){
		char vbuf[BLOCKSIZ],*oval;
		oval=vbuf;
		if(rec->sop>BLOCKSIZ) oval=malloc(rec->sop);
		off=*((uint64*)((char*)rec+BRHEADSIZ));
		bgetoverflowpage(db,off,0,oval,rec->sop);
		DEBUG(printf("%s:vsiz=%d oval=%s\n",__func__,rec->sop,oval));
		uint64 vdoff=putvdup(db,0,oval,rec->sop);
		vdoff=putvdup(db,vdoff,val,vsiz);
		dellfval(db,lf,ipos);
		rec=(struct brec*)((char*)lf+lf->index[ipos]);
		rec->sop=vdoff;
		rec->fap=B_VDUP;
		if(oval!=vbuf) free(oval);
	}else if(flag==B_PKOFLOW){
		char vbuf[BLOCKSIZ],*oval;
		oval=vbuf;
		if(rec->sop>BLOCKSIZ) oval=malloc(rec->sop);
		int pksiz=rec->ksiz-(db->osiz-BRHEADSIZ-sizeof(uint64));
		uint64 off=*((uint64*)((char*)rec+BRHEADSIZ));
		bgetoverflowpage(db,off,pksiz,oval,rec->sop);
		uint64 vdoff=putvdup(db,0,oval,rec->sop);
		vdoff=putvdup(db,vdoff,val,vsiz);
		dellfval(db,lf,ipos);
		rec=(struct brec*)((char*)lf+lf->index[ipos]);
		rec->sop=vdoff;
		rec->fap|=B_VDUP;
		DEBUG(printf("%s:PKOFLOW:pksiz=%d vdoff=%d oval=%s\n",__func__,pksiz,vdoff,oval));
		if(oval!=vbuf) free(oval);
	}else{
		printf("%s:error:fap=%x\n",__func__,rec->fap);
		exit(1);
	}
	return true;
}
int dellfval(struct bdb *db,struct blf *lf,int ipos){
    struct brec *rec=(struct brec*)((char*)lf+lf->index[ipos]),*trec;
	int rsiz,i,tsiz,flag=rec->fap&B_FMASK,inum=(lf->lower-BLHEADSIZ)/sizeof(int16);
	if(flag==0){
	    char buf[BLOCKSIZ];
	    struct blf *bl=(struct blf*)buf;
	    if(db->Bpgsiz>BLOCKSIZ) bl=malloc(db->Bpgsiz);
		DEBUG(printf("%s:inum=%d ipos=%d\n",__func__,inum,ipos));
	    bl->upper=db->Bpgsiz;
 	    bl->lower=BLHEADSIZ;
	    for(i=0;i<inum;i++){
	        if(i!=ipos){
		        trec=(struct brec*)((char*)lf+lf->index[i]);
			    tsiz=brecsiz(db,trec);
			    bl->upper-=tsiz;
			    memcpy((char*)bl+bl->upper,trec,tsiz);
			    bl->index[i]=bl->upper;
		    }
	    }
		rsiz=BRHEADSIZ+rec->ksiz;
	    lf->upper=db->Bpgsiz-rsiz;
	    memmove((char*)lf+lf->upper,rec,rsiz);
		lf->index[ipos]=lf->upper;
		rec=(struct brec*)((char*)lf+lf->index[ipos]);
		rec->sop=0;
		lf->upper-=(db->Bpgsiz-bl->upper);
		memcpy((char*)lf+lf->upper,(char*)bl+bl->upper,db->Bpgsiz-bl->upper);
		for(i=0;i<inum;i++)
		    if(i!=ipos) lf->index[i]=bl->index[i]-rsiz;
		if((char*)bl!=buf) free(bl);
	}else if(flag==B_VOFLOW){
	    char buf[BLOCKSIZ];
	    struct blf *bl=(struct blf*)buf;
		bdeloverflowpage(db,*((uint64*)((char*)rec+BRHEADSIZ)),0,rec->sop);
	    if(db->Bpgsiz>BLOCKSIZ) bl=malloc(db->Bpgsiz);
	    bl->upper=db->Bpgsiz;
	    for(i=0;i<inum;i++){
	        if(i!=ipos){
		        trec=(struct brec*)((char*)lf+lf->index[i]);
			    tsiz=brecsiz(db,trec);
			    bl->upper-=tsiz;
			    memcpy((char*)bl+bl->upper,trec,tsiz);
			    bl->index[i]=bl->upper;
		    }
	    }
	    rec->fap&=~B_VOFLOW;
		rec->sop=0;
	    lf->upper=db->Bpgsiz-rec->ksiz;
		memmove((char*)lf+lf->upper,(char*)rec+BRHEADSIZ+sizeof(uint64),rec->ksiz);
		lf->upper-=BRHEADSIZ;
		memmove((char*)lf+lf->upper,rec,BRHEADSIZ);
		rsiz=BRHEADSIZ+rec->ksiz;
		lf->index[ipos]=lf->upper;
		lf->upper-=db->Bpgsiz-bl->upper;
		memcpy((char*)lf+lf->upper,(char*)bl+bl->upper,db->Bpgsiz-bl->upper);
		for(i=0;i<inum;i++)
		    if(i!=ipos) lf->index[i]=bl->index[i]-rsiz;
		if((char*)bl!=buf) free(bl);
	}else if(flag==B_PKOFLOW){
	    int pksiz=rec->ksiz-(db->osiz-BRHEADSIZ-sizeof(uint64));
	    if(needdeloverflowpage(db,pksiz,rec->sop)){
		    uint64 off=*((uint64*)((char*)rec+BRHEADSIZ));
			bdeloverflowpage(db,off,pksiz,rec->sop);
		}
		rec->sop=0;
	}else{printf("%s:error!fap=%x\n",__func__,rec->fap);exit(1);}
	return true;
}
int needdeloverflowpage(struct bdb *db,int ksiz,int vsiz){
    int rem=ksiz+vsiz;
	rem%=db->Bpgsiz-OPHEADSIZ;
	if(rem>vsiz) return false;
	return true;
}
uint64 putvdup(struct bdb *db,uint64 off,char *val,int vsiz){
    char buf[BLOCKSIZ];
	struct vdpg *p=(struct vdpg*)buf;
	if(off==0){
	    off=bgetfpg(db);
		DEBUG(printf("%s:create a vdup page:%d vsiz=%d VP_OSIZ=%d\n",__func__,off,vsiz,VP_OSIZ));
		if(db->Bpgsiz>BLOCKSIZ) p=malloc(db->Bpgsiz);
		p->lower=VDHEADSIZ+sizeof(int16);
		p->nextpg=p->prepg=0;
		if(vsiz>VP_OSIZ){
		    uint64 oof=bwriteoverflowpage(db,NULL,0,val,vsiz);
			p->upper=db->Bpgsiz-MINVSIZ;
			char *ptr=(char*)p+p->upper;
			*((int32*)ptr)=vsiz;
			ptr+=sizeof(int32);
			*((uint64*)ptr)=oof;
			p->index[0]=p->upper;
			addpage(db->lcc,off,(char*)p,F_DIRTY);
			//seekwrite(db->fd,(char*)p,db->Bpgsiz,off);
			if((char*)p!=buf) free(p);
			return off;
		}else{
		    p->upper=db->Bpgsiz-vsiz-sizeof(int32);
			p->index[0]=p->upper;
			char *ptr=(char*)p+p->upper;
			*((int32*)ptr)=vsiz;
			ptr+=sizeof(int32);
			memcpy(ptr,val,vsiz);
			addpage(db->lcc,off,(char*)p,F_DIRTY);
			//seekwrite(db->fd,(char*)p,db->Bpgsiz,off);
			if((char*)p!=buf) free(p);
			return off;
		}
	}else{
	    p=(struct vdpg*)getpage(db->lcc,off,F_WRITE);
		uint64 oof;
		char *nval=val;
		int nvsiz=vsiz;
		if(vsiz>VP_OSIZ){
		    oof=bwriteoverflowpage(db,NULL,0,val,vsiz);
		    nval=(char*)&oof;
		    nvsiz=sizeof(uint64);
		}
		DEBUG(printf("%s:vdpg=%d vsiz=%d nvsiz=%d remsiz=%d\n",__func__,off,vsiz,nvsiz,p->upper-p->lower));
		if(nvsiz+sizeof(int32)>p->upper-p->lower-sizeof(int16)){
		    uint64 noff=bgetfpg(db);
			char buf[BLOCKSIZ];
			struct vdpg *np=(struct vdpg*)buf;
			if(db->Bpgsiz>BLOCKSIZ) np=malloc(db->Bpgsiz);
			np->nextpg=off;
			np->prepg=0;
			p->prepg=noff;
			putpage(db->lcc,(char*)p,F_DIRTY);
			np->upper=db->Bpgsiz-nvsiz;
			np->lower=VDHEADSIZ+sizeof(int16);
			np->index[0]=np->upper;
			char *ptr=(char*)np+np->upper;
			*((int32*)ptr)=vsiz;
			ptr+=sizeof(int32);
			memcpy(ptr,nval,nvsiz);
			addpage(db->lcc,noff,(char*)np,F_DIRTY);
			if((char*)np!=buf) free(np);
			return noff;
		}else{
		    char *ptr=p->lower+(char*)p;
			p->upper-=nvsiz+sizeof(int32);
			*((int16*)ptr)=p->upper;
			p->lower+=sizeof(int16);
			ptr=(char*)p+p->upper;
			*((int32*)ptr)=vsiz;
			ptr+=sizeof(int32);
			memcpy(ptr,nval,nvsiz);
			putpage(db->lcc,(char*)p,F_DIRTY);
			return off;
		}
		
	}
}
int bputrec(struct bdb *db,struct blf *lf,struct brec *rec,int rsiz){
    int ipos=db->stack[db->snum-1].idxoff;
	int rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
	DEBUG(printf("%s:ipos=%d rnum=%d remsiz=%d rsiz=%d dir=%d snum=%d\n",__func__,ipos,rnum,lf->upper-lf->lower,rsiz,db->dir,db->snum));
	if(lf->upper-lf->lower>=rsiz+sizeof(int16)){
	    lf->lower+=sizeof(int16);
		lf->upper-=rsiz;
		memmove(lf->index+ipos+1,lf->index+ipos,(rnum-ipos)*sizeof(int16));
		lf->index[ipos]=lf->upper;
		memcpy((char*)lf+lf->upper,rec,rsiz);
		putpage(db->lcc,(char*)lf,F_DIRTY);
		pthread_mutex_lock(&db->iter.lock);
		if((db->iter.flag&BI_INUSE)&&db->iter.offset==db->stack[db->snum-1].offset
		   &&db->iter.idxoff>=ipos)
		    db->iter.idxoff++;
		pthread_mutex_unlock(&db->iter.lock);
	}else{
	    char buf[BLOCKSIZ];
		char ibuf[BLOCKSIZ];
		struct blf *rlf;
		struct bidx *idx;
		int isiz=BLOCKSIZ;
		if(db->Bpgsiz>BLOCKSIZ){
			rlf=malloc(db->Bpgsiz);
		}else
		    rlf=(struct blf*)buf;
		if(ipos==rnum&&db->dir>1){
		    rlf->upper=db->Bpgsiz-rsiz;
		    rlf->lower=BLHEADSIZ+sizeof(int16);
		    rlf->index[0]=rlf->upper;
		    memcpy((char*)rlf+rlf->upper,rec,rsiz);
			idx=bidxcreate(db,rec,ibuf,&isiz);
			if(db->snum==1){
			    uint64 loff=bgetfpg(db);
			    uint64 roff=bgetfpg(db);
				lf->next=roff;
				rlf->pre=loff;
				rlf->next=0;
			    DEBUG(printf("%s:left pgno=%d create new right pgno=%d no split\n",__func__,loff,roff));
			    addpage(db->lcc,loff,(char*)lf,F_DIRTY);
			    struct bpg *p=(struct bpg*)lf;
			    p->lower=BPHEADSIZ+sizeof(int16);
			    p->upper=db->Bpgsiz-isiz;
			    p->index[0]=p->upper;
			    p->left=BLEAF(loff);
			    idx->offset=BLEAF(roff);
			    memcpy((char*)p+p->upper,idx,isiz);
			    void *rootp=removenode(db->lcc,(char*)p);
			    insertnode(db->pcc,rootp,F_DIRTY);
			    addpage(db->lcc,roff,(char*)rlf,F_DIRTY);
			    db->Blmost=loff;
			    db->Brmost=roff;
			    db->lastpg=roff;
			    db->snum=0;
			    db->Broot=OFFSET(db->Broot);
			}else{
			    uint64 off=bgetfpg(db);
			    if(lf->next!=0){
			        struct blf* rrl=(struct blf*)getpage(db->lcc,lf->next,F_WRITE);
				    rrl->pre=off;
				    putpage(db->lcc,(char*)rrl,F_DIRTY);
			    }
			    rlf->next=lf->next;
			    rlf->pre=db->stack[db->snum-1].offset;
			    DEBUG(printf("%s:new right leaf=%d orig lf->next=%d\n",__func__,off,lf->next));
			    lf->next=off;
			    if(db->Brmost==OFFSET(db->stack[db->snum-1].offset)) db->Brmost=off;
			    addpage(db->lcc,off,(char*)rlf,F_DIRTY);
			    putpage(db->lcc,(char*)lf,F_DIRTY);
			    idx->offset=BLEAF(off);
		        bputidx(db,NULL,db->snum-1,idx,isiz);
			    db->lastpg=off;
				db->snum=0;
			}
		}else if(ipos==0&&db->dir<-1){
		    rlf->upper=db->Bpgsiz-rsiz;
		    rlf->lower=BLHEADSIZ+sizeof(int16);
		    rlf->index[0]=rlf->upper;
		    memcpy((char*)rlf+rlf->upper,rec,rsiz);
			struct brec *trec=(struct brec*)((char*)lf+lf->index[0]);
			idx=bidxcreate(db,trec,ibuf,&isiz);
			if(db->snum==1){
			    uint64 loff=bgetfpg(db);
			    uint64 roff=bgetfpg(db);
				lf->next=0;
				lf->pre=loff;
				rlf->pre=0;
				rlf->next=roff;
			    DEBUG(printf("%s:create new left pgno=%d right pgno=%d\n",__func__,loff,roff));
			    addpage(db->lcc,loff,(char*)rlf,F_DIRTY);
				addpage(db->lcc,roff,(char*)lf,F_DIRTY);
			    struct bpg *p=(struct bpg*)lf;
			    p->lower=BPHEADSIZ+sizeof(int16);
			    p->upper=db->Bpgsiz-isiz;
			    p->index[0]=p->upper;
			    p->left=BLEAF(loff);
			    idx->offset=BLEAF(roff);
			    memcpy((char*)p+p->upper,idx,isiz);
			    void *rootp=removenode(db->lcc,(char*)p);
			    insertnode(db->pcc,rootp,F_DIRTY);
			    //addpage(db->lcc,roff,(char*)rlf,F_DIRTY);
			    db->Blmost=loff;
			    db->Brmost=roff;
			    db->lastpg=loff;
			    db->snum=0;
			    db->Broot=OFFSET(db->Broot);
			}else{
			    uint64 off=bgetfpg(db);
			    if(lf->pre!=0){
			        struct blf* ll=(struct blf*)getpage(db->lcc,lf->pre,F_WRITE);
				    ll->next=off;
				    putpage(db->lcc,(char*)ll,F_DIRTY);
			    }
			    rlf->pre=lf->pre;
			    rlf->next=db->stack[db->snum-1].offset;
				DEBUG(printf("%s:new left leaf=%d orig lf->pre=%d\n",__func__,off,lf->pre));
			    lf->pre=off;
			    if(db->Blmost==OFFSET(db->stack[db->snum-1].offset)) db->Blmost=off;
			    addpage(db->lcc,off,(char*)rlf,F_DIRTY);
			    putpage(db->lcc,(char*)lf,F_DIRTY);
			    idx->offset=BLEAF(db->stack[db->snum-1].offset);
				struct bpg *p=(struct bpg*)getpage(db->pcc,db->stack[db->snum-2].offset,F_WRITE);
				p->left=BLEAF(off);
				putpage(db->pcc,(char*)p,F_DIRTY);
		        bputidx(db,p,db->snum-1,idx,isiz);
			    db->lastpg=off;
				db->snum=0;
			}
		}else{
		    pthread_mutex_lock(&db->iter.lock);
		    if((db->iter.flag&BI_INUSE)&&db->iter.offset==db->stack[db->snum-1].offset
			   &&db->iter.idxoff>=ipos)
			    db->iter.idxoff++;
		    int lor=bsplitleaf(db,lf,ipos,rec,rsiz,rlf);
		    struct brec *trec=(struct brec*)((char*)rlf+rlf->index[0]);
	        idx=bidxcreate(db,trec,ibuf,&isiz);
		    if(db->snum==1){
		        uint64 loff=bgetfpg(db);
			    uint64 roff=bgetfpg(db);
			    lf->next=roff;
			    rlf->next=0;
			    rlf->pre=loff;
			    DEBUG(printf("%s:left pgno=%d right pgno=%d\n",__func__,loff,roff));
				addpage(db->lcc,loff,(char*)lf,F_DIRTY);
				if((db->iter.flag&BI_INUSE)&&(db->iter.offset==OFFSET(db->Broot))){
				    int lnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
					if(lnum<=db->iter.idxoff){
					    db->iter.idxoff-=lnum;//
					    db->iter.offset=roff;
					}
				}
				pthread_mutex_unlock(&db->iter.lock);
			    struct bpg *p=(struct bpg*)lf;
			    p->lower=BPHEADSIZ+sizeof(int16);
			    p->upper=db->Bpgsiz-isiz;
			    p->index[0]=p->upper;
			    p->left=BLEAF(loff);
			    idx->offset=BLEAF(roff);
			    memcpy((char*)p+p->upper,idx,isiz);
			    void *rootp=removenode(db->lcc,(char*)p);
			    insertnode(db->pcc,rootp,F_DIRTY);
			    addpage(db->lcc,roff,(char*)rlf,F_DIRTY);
			    db->Blmost=loff;
			    db->Brmost=roff;
			    db->lastpg=lor==1?loff:roff;
			    db->snum=0;
			    db->Broot=OFFSET(db->Broot);
				
		    }else{
		        uint64 off=bgetfpg(db);
			    if(lf->next!=0){
			        struct blf* rrl=(struct blf*)getpage(db->lcc,lf->next,F_WRITE);
				    rrl->pre=off;
				    putpage(db->lcc,(char*)rrl,F_DIRTY);
			    }
			    rlf->next=lf->next;
			    rlf->pre=db->stack[db->snum-1].offset;
			    lf->next=off;
			    DEBUG(printf("%s:new right leaf=%d db->Brmost=%d db->stack[%d].offset=%d\n",__func__,off,db->Brmost,db->snum-1,db->stack[db->snum-1].offset));
			    if(db->Brmost==OFFSET(db->stack[db->snum-1].offset)) db->Brmost=off;
				if((db->iter.flag&BI_INUSE)&&(db->iter.offset==OFFSET(db->stack[db->snum-1].offset))){
				    int lnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
					if(lnum<=db->iter.idxoff){
					    db->iter.idxoff-=lnum;//
					    db->iter.offset=off;
					}
				}
				pthread_mutex_unlock(&db->iter.lock);
			    addpage(db->lcc,off,(char*)rlf,F_DIRTY);
			    putpage(db->lcc,(char*)lf,F_DIRTY);
			    idx->offset=BLEAF(off);
		        bputidx(db,NULL,db->snum-1,idx,isiz);
			    if(lor==2){
			        db->lastpg=off;
				    db->snum=0;
			    }
		    }
		}
		if((char*)rlf!=buf) free(rlf);
		if((char*)idx!=ibuf) free(idx);
	}
	return true;
}
int bsplitleaf(struct bdb *db,struct blf *lf,int ipos,struct brec *rec,int rsiz,struct blf* rlf){
    int half=(db->Bpgsiz-BLHEADSIZ)>>1;
	int rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
	int i,isplit,siz=0,tsiz,ret;
	struct brec *trec;
	char buf[BLOCKSIZ];
	struct blf *llf=(struct blf*)buf;
	if(db->Bpgsiz>BLOCKSIZ) llf=malloc(db->Bpgsiz);
	for(isplit=rnum-1;isplit>=0;isplit--){
	    trec=(struct brec*)((char*)lf+lf->index[isplit]);
		DEBUG(printf("%s:%d rec siz=%d\n",__func__,isplit,brecsiz(db,trec)));
		siz+=(tsiz=brecsiz(db,trec))+sizeof(int16);
		if(siz>=half){
		    if(siz>half&&isplit<ipos&&rnum-isplit>=db->Bminkey-1){
			   isplit++;
			   siz-=tsiz;
			}
			break;
		}
	}
	DEBUG(printf("%s:ipos=%d isplit=%d rnum=%d siz=%d\n",__func__,ipos,isplit,rnum,siz));
	rlf->lower=BLHEADSIZ;
	rlf->upper=db->Bpgsiz;
	for(i=isplit;i<rnum;i++){
	    trec=(struct brec*)((char*)lf+lf->index[i]);
		tsiz=brecsiz(db,trec);
		rlf->upper-=tsiz;
		memcpy((char*)rlf+rlf->upper,trec,tsiz);
		rlf->index[i-isplit]=rlf->upper;
		rlf->lower+=sizeof(int16);
	}
	llf->lower=BLHEADSIZ;
	llf->upper=db->Bpgsiz;
	for(i=0;i<isplit;i++){
	    trec=(struct brec*)((char*)lf+lf->index[i]);
		tsiz=brecsiz(db,trec);
		llf->upper-=tsiz;
		memcpy((char*)llf+llf->upper,trec,tsiz);
		llf->index[i]=llf->upper;
	}
	memcpy(lf->index,llf->index,i*sizeof(int16));
	lf->lower=BLHEADSIZ+i*sizeof(int16);
	lf->upper=llf->upper;
	memcpy((char*)lf+lf->upper,(char*)llf+llf->upper,db->Bpgsiz-llf->upper);
	if(isplit<ipos||(isplit==ipos&&siz+rsiz+sizeof(int16)<=db->Bpgsiz-BLHEADSIZ)){
		int16 nipos=ipos-isplit;
		memmove(rlf->index+nipos+1,rlf->index+nipos,(rnum-isplit-nipos)*sizeof(int16));
		rlf->upper-=rsiz;
		memcpy(rlf->upper+(char*)rlf,rec,rsiz);
		rlf->index[nipos]=rlf->upper;
		rlf->lower+=sizeof(int16);
		DEBUG(printf("%s:right nipos=%d rnum=%d\n",__func__,nipos,(rlf->lower-BLHEADSIZ)/sizeof(int16)));
		ret=2;
	}else{
	    memmove(lf->index+ipos+1,lf->index+ipos,(isplit-ipos)*sizeof(int16));
		lf->upper-=rsiz;
		memcpy(lf->upper+(char*)lf,rec,rsiz);
		lf->index[ipos]=lf->upper;
		lf->lower+=sizeof(int16);
		ret=1;
	}
	if((char*)llf!=buf) free(llf);
	return ret;
}
int idxsiz(struct bdb *db,struct bidx *idx){
    int flag=idx->fap&B_FMASK;
	if(flag==0)
	    return idx->ksiz+BIHEADSIZ;
	else if(flag==B_PKOFLOW) return db->osiz;
	else{
	    printf("%s:error:fap=%x\n",__func__,idx->fap);
		exit(1);
	}
}
int bsplitpage(struct bdb *db,struct bpg* p,int ipos,struct bidx *idx,int isiz,struct bpg* rp,struct bidx** ridx,int *rsiz){
    int inum=(p->lower-BPHEADSIZ)/sizeof(int16);
	int i,siz=0,isplit,tsiz,half=(db->Bpgsiz-BPHEADSIZ)>>1;
	struct bidx *tidx;
	char lbuf[BLOCKSIZ];
	struct bpg *lp=(struct bpg*)lbuf;
	if(db->Bpgsiz>BLOCKSIZ) lp=malloc(db->Bpgsiz);
	for(isplit=0;isplit<inum;isplit++){
	    tidx=(struct bidx*)((char*)p+p->index[isplit]);
		siz+=(tsiz=idxsiz(db,tidx))+sizeof(int16);
		if(siz>=half){
		    if(siz>half&&ipos<isplit&&isplit>=db->Bminkey){
			    isplit--;
				siz-=tsiz;
			}else if(siz==half&&ipos+1>isplit&&isplit<=db->Bminkey)
			    isplit++;
		    break;
		}
	}
	rp->upper=db->Bpgsiz;
	int start=(isplit==ipos&&isplit>=db->Bminkey)?isplit:isplit+1;
	DEBUG(printf("%s:ipos=%d isplit=%d siz=%d start=%d\n",__func__,ipos,isplit,siz,start));
	for(i=start;i<inum;i++){
	    tidx=(struct bidx*)((char*)p+p->index[i]);
		tsiz=idxsiz(db,tidx);
		rp->upper-=tsiz;
		memcpy((char*)rp+rp->upper,tidx,tsiz);
		rp->index[i-start]=rp->upper;
	}
	rp->lower=BPHEADSIZ+sizeof(int16)*(inum-start);
	lp->upper=db->Bpgsiz;
	for(i=0;i<start;i++){
	    tidx=(struct bidx*)((char*)p+p->index[i]);
		tsiz=idxsiz(db,tidx);
		lp->upper-=tsiz;
		memcpy((char*)lp+lp->upper,tidx,tsiz);
		lp->index[i]=lp->upper;
	}
	p->lower=BPHEADSIZ+start*sizeof(int16);
	memcpy(p->index,lp->index,p->lower-BPHEADSIZ);
	p->upper=lp->upper;
	memcpy((char*)p+p->upper,(char*)lp+lp->upper,db->Bpgsiz-lp->upper);
	if(ipos==isplit&&isplit>=db->Bminkey){
	    if(*rsiz<isiz) *ridx=malloc(isiz);
		*rsiz=isiz;
		memcpy(*ridx,idx,isiz);
		rp->left=idx->offset;
		DEBUG(printf("%s:split self left inum=%d right inum=%d\n",__func__,(p->lower-BPHEADSIZ)/sizeof(int16),(rp->lower-BPHEADSIZ)/sizeof(int16)));
	}else{
	    tidx=(struct bidx*)(lp->index[isplit]+(char*)lp);
		tsiz=idxsiz(db,tidx);
		if(tsiz>*rsiz) *ridx=malloc(tsiz);
		*rsiz=tsiz;
		memcpy(*ridx,tidx,tsiz);
		rp->left=tidx->offset;
		p->upper+=tsiz;
		p->lower-=sizeof(int16);
		DEBUG(printf("%s:left inum=%d\n",__func__,(p->lower-BPHEADSIZ)/sizeof(int16)));
	    if(ipos>isplit){
	        int nipos=ipos-start;
	        rp->upper-=isiz;
		    memmove(rp->index+nipos+1,rp->index+nipos,(inum-start-nipos)*sizeof(int16));
		    rp->index[nipos]=rp->upper;
		    memcpy(rp->upper+(char*)rp,idx,isiz);
		    DEBUG(printf("%s:!!right:idx->offset=%d nipos=%d isiz=%d\n",__func__,idx->offset,nipos,isiz));
		    rp->lower+=sizeof(int16);
	    }
	    else{
	        p->upper-=isiz;
		    memmove(p->index+ipos+1,p->index+ipos,(start-ipos)*sizeof(int16));
		    p->index[ipos]=p->upper;
		    memcpy((char*)p+p->upper,idx,isiz);
		    p->lower+=sizeof(int16);
			DEBUG(printf("%s:!!left remsiz=%d\n",__func__,p->upper-p->lower));
	    }
	}
	if((char*)lp!=lbuf) free(lp);
	return 1;
}
int bputidx(struct bdb *db,struct bpg* p,int snum,struct bidx *idx,int isiz){
	uint64 roff;
	int16 ipos;
	int inum,rsiz,i=0;
	char ibuf[2][BLOCKSIZ];
	struct bidx *ridx,*sidx=idx;
	DEBUG(printf("%s:idx->offset=%d snum=%d\n",__func__,idx->offset,snum));
	if(p==NULL) p=(struct bpg*)getpage(db->pcc,db->stack[snum-1].offset,F_WRITE);
	while(snum>1){
	    ipos=db->stack[snum-1].idxoff;
	    inum=(p->lower-BPHEADSIZ)/sizeof(int16);
		DEBUG(printf("%s:snum=%d off=%d ipos=%d inum=%d isiz=%d remsiz=%d\n",__func__,snum,db->stack[snum-1].offset,ipos,inum,isiz,p->upper-p->lower));
	    if(p->upper-p->lower>=isiz+sizeof(int16)){
	        p->upper-=isiz;
		    memcpy((char*)p+p->upper,sidx,isiz);
		    p->lower+=sizeof(int16);
		    memmove(p->index+ipos+1,p->index+ipos,(inum-ipos)*sizeof(int16));
		    p->index[ipos]=p->upper;
			putpage(db->pcc,(char*)p,F_DIRTY);
			if(sidx!=idx&&(char*)sidx!=ibuf[i]) free(sidx);
			return true;
	    }else{
	        char buf[BLOCKSIZ];
	        struct bpg *rp=(struct bpg*)buf;
			ridx=(struct bidx*)ibuf[!i];
			rsiz=BLOCKSIZ;
		    if(db->Bpgsiz>BLOCKSIZ) rp=malloc(db->Bpgsiz);
		    bsplitpage(db,p,ipos,sidx,isiz,rp,&ridx,&rsiz);
			if(sidx!=idx&&(char*)sidx!=ibuf[i]) free(sidx);
			roff=bgetfpg(db);
			DEBUG(printf("%s:new right inner pgno=%d\n",__func__,roff));
			sidx=ridx;
			isiz=rsiz;
			i=!i;
			sidx->offset=roff;
			addpage(db->pcc,roff,(char*)rp,F_DIRTY);
			putpage(db->pcc,(char*)p,F_DIRTY);
			if((char*)rp!=buf) free(rp);
			db->snum=0;
		}
		snum--;
		p=(struct bpg*)getpage(db->pcc,db->stack[snum-1].offset,F_WRITE);
	}
	if(p->upper-p->lower<isiz){
		uint64 loff=bgetfpg(db);
		uint64 roff=bgetfpg(db);
		char rbuf[BLOCKSIZ];
		struct bpg *rp=(struct bpg*)rbuf;
		if(db->Bpgsiz>BLOCKSIZ) rp=malloc(db->Bpgsiz);
		ridx=(struct bidx*)ibuf[!i];
		rsiz=BLOCKSIZ;
		DEBUG(printf("%s:loff=%d roff=%d isiz=%d remsiz=%d ipos=%d sidx->offset=%d\n",__func__,loff,roff,isiz,p->upper-p->lower,db->stack[0].idxoff,sidx->offset));
		bsplitpage(db,p,db->stack[0].idxoff,sidx,isiz,rp,&ridx,&rsiz);
		addpage(db->pcc,loff,(char*)p,F_DIRTY);
		p->upper=db->Bpgsiz-rsiz;
		p->lower=BPHEADSIZ+sizeof(int16);
		p->index[0]=p->upper;
		ridx->offset=roff;
		memcpy((char*)p+p->upper,ridx,rsiz);
		addpage(db->pcc,roff,(char*)rp,F_DIRTY);
		p->left=loff;
		putpage(db->pcc,(char*)p,F_DIRTY);
		if((char*)ridx!=ibuf[!i]) free(ridx);
		if((char*)rp!=rbuf) free(rp);
		db->snum=0;
    }else{
	    inum=(p->lower-BPHEADSIZ)/sizeof(int16);
	    p->upper-=isiz;
		p->lower+=sizeof(int16);
		ipos=db->stack[0].idxoff;
		DEBUG(printf("%s:inum=%d ipos=%d p->left=%d sidx->offset=%d\n",__func__,inum,ipos,p->left,sidx->offset));
		memmove(p->index+ipos+1,p->index+ipos,sizeof(int16)*(inum-ipos));
		p->index[ipos]=p->upper;
		memcpy(p->upper+(char*)p,sidx,isiz);
	}
	if(sidx!=idx&&(char*)sidx!=ibuf[i]) free(sidx);
	return true;
}
int bclose(struct bdb *db){
    pthread_rwlock_wrlock(&db->rwlock);
    cachedel(db->pcc);
	cachedel(db->lcc);
	if(db->Bnfpg>=(db->Bmaxpg>>1))
	    bshrinkpage(db);
	DEBUG(printf("%s:db->Bmaxpg=%d\n",__func__,db->Bmaxpg));
	ftruncate(db->fd,db->Bmaxpg);
	if(msync(db->head,512,MS_SYNC)<0) fprintf(stderr,"msync:%s\n",strerror(errno));
	DEBUG(printf("%s:idxname=%s\n",__func__,(char*)db->head+BDB_HEAD_SIZE+sizeof(int32)+sizeof(int32)));
	munmap(db->head,512);
	fsync(db->fd);
	close(db->fd);
	if(db->stack!=NULL)
	    free(db->stack);
	free(db->dname);
	pthread_mutex_unlock(&db->iter.lock);
	pthread_rwlock_unlock(&db->rwlock);
	pthread_rwlock_destroy(&db->rwlock);
	free(db);
	return true;
}
int readsearch(struct bdb *db,char *key,int ksiz,struct bc *cursor){
    struct bpg *pg;
	struct blf *lf;
	struct bidx *idx;
	int ipos,hit;
	uint64 off=db->Broot;
	if(off==0) return 0;
	while(!ISBLEAF(off)){
	    pg=(struct bpg*)getpage(db->pcc,off,F_READ);
		ipos=pgsearch(db,pg,key,ksiz);
		if(ipos<1) off=pg->left;
		else{
		    idx=(struct bidx*)(pg->index[ipos-1]+(char*)pg);
			off=idx->offset;	
		}
		putpage(db->pcc,(char*)pg,0);
	}
	lf=(struct blf*)getpage(db->lcc,OFFSET(off),F_READ);
	ipos=lfsearch(db,lf,key,ksiz,&hit);
	if(hit)
	    BC(cursor,off,ipos,lf);
	else putpage(db->lcc,(char*)lf,0);
	return hit;
}
int bget(struct bdb *db,struct bstring *key,struct strlist **list){
    struct blf *lf;
	struct bc cursor;
	pthread_rwlock_rdlock(&db->rwlock);
	if(!readsearch(db,key->array,key->siz,&cursor)){
	    pthread_rwlock_unlock(&db->rwlock);
	    return false;
	}
	int ipos=cursor.idxoff;
	lf=cursor.p;
	struct brec *rec=(struct brec*)((char*)lf+lf->index[ipos]);
	int flag=rec->fap&B_FMASK;
	if(flag==0){
	    struct bstring *val=strcreate((char*)rec+BRHEADSIZ+rec->ksiz,rec->sop);
		if(*list==NULL) *list=listcreate(1);
		listadd(*list,val);
	}else if(flag==B_VOFLOW){
	    uint64 off=*((uint64*)((char*)rec+BRHEADSIZ));
		struct bstring *val=strnew(rec->sop);
		bgetoverflowpage(db,off,0,val->array,rec->sop);
		val->siz=rec->sop;
		if(*list==NULL) *list=listcreate(1);
		listadd(*list,val);
	}else if(flag&B_VDUP){
	    uint64 off=rec->sop;
		struct vdpg *vp;
		int vnum,i,vsiz;
		char *ptr;
		struct bstring *val;
		if(*list==NULL) *list=listcreate(0);
		while(off!=0){
			vp=(struct vdpg*)getpage(db->lcc,off,F_READ);
			vnum=(vp->lower-VDHEADSIZ)/sizeof(int16);
			DEBUG(printf("%s:vdup pgno=%d vnum=%d\n",__func__,off,vnum));
			for(i=0;i<vnum;i++){
			    ptr=vp->index[i]+(char*)vp;
				vsiz=*((int32*)ptr);
				ptr+=sizeof(int32);
				val=strnew(vsiz);
				DEBUG(printf("%s:val %d siz=%d\n",__func__,i,vsiz));
				if(vsiz>VP_OSIZ){
				    uint64 oof=*((uint64*)ptr);
					bgetoverflowpage(db,oof,0,val->array,vsiz);
					val->siz=vsiz;
				}else{
				    memcpy(val->array,ptr,vsiz);
					val->siz=vsiz;
				}
				listadd(*list,val);
			}
			putpage(db->lcc,(char*)vp,0);
			off=vp->nextpg;
		}
	}else if(flag==B_PKOFLOW){
	    int pksiz=rec->ksiz-(db->osiz-BRHEADSIZ-sizeof(int64));
		uint64 off=*((uint64*)((char*)rec+BRHEADSIZ));
		DEBUG(printf("%s:pkpgno=%d ksiz=%d\n",__func__,off,rec->sop));
		struct bstring *val=strnew(rec->sop);
		bgetoverflowpage(db,off,pksiz,val->array,rec->sop);
		val->siz=rec->sop;
		if(*list==NULL) *list=listcreate(1);
		listadd(*list,val);
	}else{
	    printf("%s:error:fap=%x\n",__func__,rec->fap);
		exit(1);
	}
	putpage(db->lcc,(char*)lf,0);
	pthread_rwlock_unlock(&db->rwlock);
	return true;
}
int bgettohash(struct bdb *db,char *key, int ksiz,struct harray **ha){
    struct blf *lf;
	struct bc cursor;
	pthread_rwlock_rdlock(&db->rwlock);
	printf("%s:key=%s\n",__func__,key);
	if(!readsearch(db,key,ksiz,&cursor)){
	    pthread_rwlock_unlock(&db->rwlock);
	    return false;
	}
	if(*ha==NULL) *ha=hashcreate(0);
	int ipos=cursor.idxoff;
	lf=cursor.p;
	struct brec *rec=(struct brec*)((char*)lf+lf->index[ipos]);
	int flag=rec->fap&B_FMASK;
	if(flag==0){
		hashput(*ha,(char*)rec+BRHEADSIZ+rec->ksiz,rec->sop,NULL);
	}else if(flag==B_VOFLOW){
	    uint64 off=*((uint64*)((char*)rec+BRHEADSIZ));
		struct bstring *val=strnew(rec->sop);
		bgetoverflowpage(db,off,0,val->array,rec->sop);
		val->siz=rec->sop;
		hashputstr(*ha,val,NULL);
	}else if(flag&B_VDUP){
	    uint64 off=rec->sop;
		struct vdpg *vp;
		int vnum,i,vsiz;
		char *ptr;
		struct bstring *val;
		while(off!=0){
			vp=(struct vdpg*)getpage(db->lcc,off,F_READ);
			vnum=(vp->lower-VDHEADSIZ)/sizeof(int16);
			DEBUG(printf("%s:vdup pgno=%d vnum=%d\n",__func__,off,vnum));
			for(i=0;i<vnum;i++){
			    ptr=vp->index[i]+(char*)vp;
				vsiz=*((int32*)ptr);
				ptr+=sizeof(int32);
				val=strnew(vsiz);
				DEBUG(printf("%s:val %d siz=%d\n",__func__,i,vsiz));
				if(vsiz>VP_OSIZ){
				    uint64 oof=*((uint64*)ptr);
					bgetoverflowpage(db,oof,0,val->array,vsiz);
					val->siz=vsiz;
				}else{
				    memcpy(val->array,ptr,vsiz);
					val->siz=vsiz;
				}
				//listadd(list,val);
				hashputstr(*ha,val,NULL);
			}
			putpage(db->lcc,(char*)vp,0);
			off=vp->nextpg;
		}
	}else if(flag==B_PKOFLOW){
	    int pksiz=rec->ksiz-(db->osiz-BRHEADSIZ-sizeof(int64));
		uint64 off=*((uint64*)((char*)rec+BRHEADSIZ));
		DEBUG(printf("%s:pkpgno=%d ksiz=%d\n",__func__,off,rec->sop));
		struct bstring *val=strnew(rec->sop);
		bgetoverflowpage(db,off,pksiz,val->array,rec->sop);
		val->siz=rec->sop;
		hashputstr(*ha,val,NULL);
	}else{
	    printf("%s:error:fap=%x\n",__func__,rec->fap);
		exit(1);
	}
	putpage(db->lcc,(char*)lf,0);
	pthread_rwlock_unlock(&db->rwlock);
	return true;
}
int berase(struct bdb *db,char *key,int ksiz,char *val,int vsiz){
    struct blf *lf;
	int hit=0,rnum,flag;
	struct brec *rec;
	pthread_rwlock_wrlock(&db->rwlock);
	if(db->Broot==0||(lf=rsearch(db,key,ksiz,&hit))==NULL||hit==0){
	    if(lf!=NULL) putpage(db->lcc,(char*)lf,0);
		pthread_rwlock_unlock(&db->rwlock);
		return false;
	}
	rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
	rec=(struct brec*)((char*)lf+lf->index[db->stack[db->snum-1].idxoff]);
	flag=rec->fap&B_FMASK;
	if(!(flag&B_VDUP)){
	    bdelrec(db,lf,db->snum);
		pthread_rwlock_unlock(&db->rwlock);
		return true;
	}
	uint64 off=rec->sop;
	int ret;
	DEBUG(printf("%s:del vdup pg=%d\n",__func__,off));
	if(val!=NULL&&vsiz!=0){//delete the key and value exactly..
	    ret=delspecialval(db,off,val,vsiz,(uint64*)&rec->sop);
		if(ret==0){//no more val,del record
		    bdelrec(db,lf,db->snum);
		}
		else if(ret==-1){
            putpage(db->lcc,(char*)lf,0);
            pthread_rwlock_unlock(&db->rwlock);			
		    return false;//not find the special value and key..
		}
	}else{//delete all the values and key
	    deldupval(db,off);
		bdelrec(db,lf,db->snum);
	}
	pthread_rwlock_unlock(&db->rwlock);
	return true;
}
int bdelrec(struct bdb *db,struct blf *lf,int snum){
    int rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
	int ipos=db->stack[snum-1].idxoff;
	struct brec *rec=(struct brec*)((char*)lf+lf->index[ipos]);
	int flag=rec->fap&B_FMASK;
	DEBUG(printf("%s:snum=%d ipos=%d fap=%x\n",__func__,snum,ipos,rec->fap));
	if(flag&B_PKOFLOW){//del overflowpage,if the overflowpage is referenced by inner page,just decrease its reference
	    if(!(flag&B_VDUP))
		    bdeloverflowpage(db,*((uint64*)((char*)rec+BRHEADSIZ)),rec->ksiz-(db->osiz-BRHEADSIZ-sizeof(uint64)),rec->sop);
		delopg(db,*((uint64*)((char*)rec+BRHEADSIZ)));
	}else if(flag&B_VOFLOW){
	    bdeloverflowpage(db,*((uint64*)((char*)rec+BRHEADSIZ)),0,rec->sop);
	}
	if(rnum<=1){//delete whole page..
	    if(lf->next!=0){//make the leaf page link looks well
	        struct blf *bl=(struct blf*)getpage(db->lcc,lf->next,F_WRITE);
		    bl->pre=lf->pre;
		    putpage(db->lcc,(char*)bl,F_DIRTY);
		}
		if(lf->pre!=0){
		    struct blf *bl=(struct blf*)getpage(db->lcc,lf->pre,F_WRITE);
			bl->next=lf->next;
			putpage(db->lcc,(char*)bl,F_DIRTY);
		}
		pthread_mutex_lock(&db->iter.lock);
		if((db->iter.flag&BI_INUSE)&&db->iter.offset==db->stack[snum-1].offset){
		    if(db->iter.flag&BI_BACKWORD){
		        db->iter.offset=lf->pre;
				db->iter.idxoff=INT16_MAX;
			}else{
			    db->iter.offset=lf->next;
				db->iter.idxoff=0;
				db->iter.voffset=0;
			}
		}
		pthread_mutex_unlock(&db->iter.lock);
		if(OFFSET(db->stack[snum-1].offset)==db->Blmost) db->Blmost=lf->next;
		if(OFFSET(db->stack[snum-1].offset)==db->Brmost) db->Brmost=lf->pre;
	    bfreepg(db,OFFSET(db->stack[snum-1].offset));
		DEBUG(printf("%s:free pgno=%d\n",__func__,db->stack[snum-1].offset));
		delpage(db->lcc,(char*)lf);
		if(snum==1){
            db->Broot=0;
			return true;
		}		
	    bdelidx(db,snum-1);
		db->snum=0;
		
	}else{//just delete the record
		struct brec *trec;
		int i,k=0,tsiz;
		char buf[BLOCKSIZ];
		struct blf *bl=(struct blf*)buf;
		if(db->Bpgsiz>BLOCKSIZ) bl=malloc(db->Bpgsiz);
		bl->upper=db->Bpgsiz;
		for(i=0;i<rnum;i++){
		    if(i!=ipos){
			    trec=(struct brec*)((char*)lf+lf->index[i]);
				tsiz=brecsiz(db,trec);
				bl->upper-=tsiz;
				memcpy((char*)bl+bl->upper,trec,tsiz);
				bl->index[k++]=bl->upper;
			}
		}
		lf->upper=bl->upper;
		lf->lower-=sizeof(int16);
		memcpy((char*)lf+lf->upper,(char*)bl+bl->upper,db->Bpgsiz-bl->upper);
		memcpy(lf->index,bl->index,lf->lower-BLHEADSIZ);
		pthread_mutex_lock(&db->iter.lock);
		if((db->iter.flag&BI_INUSE)&&db->iter.offset==db->stack[db->snum-1].offset
		    &&db->iter.idxoff>=ipos&&db->iter.flag&BI_BACKWORD){
			    db->iter.idxoff--;
				db->iter.voffset=0;
			}
		pthread_mutex_unlock(&db->iter.lock);
		putpage(db->lcc,(char*)lf,F_DIRTY);
		if((char*)bl!=buf) free(bl);
	}
	return true;
}
int bdelidx(struct bdb *db,int snum){
    struct bpg* p=(struct bpg*)getpage(db->pcc,db->stack[snum-1].offset,F_WRITE);
	int inum=(p->lower-BPHEADSIZ)/sizeof(int16);
	int ipos,isiz;
	struct bidx *idx;
	while(inum==0&&snum>1){//if the page has no record,delete the whole page
	    bfreepg(db,db->stack[snum-1].offset);
		DEBUG(printf("%s:free pgno=%d\n",__func__,db->stack[snum-1].offset));
		delpage(db->pcc,(char*)p);
		snum--;
		p=(struct bpg*)getpage(db->pcc,db->stack[snum-1].offset,F_WRITE);
		inum=(p->lower-BPHEADSIZ)/sizeof(int16);
	}
	if(snum==1&&inum==0){//delete root..
	    delpage(db->pcc,(char*)p);
	    bfreepg(db,db->Broot);
		db->Broot=0;
		return true;
	}
	ipos=db->stack[snum-1].idxoff;
	if(ipos==0){//delete page left,then delete first index..
	    char buf[BLOCKSIZ];
		struct bpg* bp=(struct bpg*)buf;
		int i;
		idx=(struct bidx*)((char*)p+p->index[0]);
		if(idx->fap&B_PKOFLOW){//decrease the reference
		    delopg(db,*((uint64*)((char*)idx+BIHEADSIZ)));
		}
		if(db->Bpgsiz>BLOCKSIZ) bp=malloc(db->Bpgsiz);
		p->left=((struct bidx*)((char*)p+p->index[0]))->offset;
		bp->upper=db->Bpgsiz;
		for(i=1;i<inum;i++){
		    idx=(struct bidx*)((char*)p+p->index[i]);
			isiz=idxsiz(db,idx);
			bp->upper-=isiz;
			memcpy((char*)bp+bp->upper,idx,isiz);
			bp->index[i-1]=bp->upper;
		}
		p->upper=bp->upper;
		memcpy((char*)p+p->upper,(char*)bp+bp->upper,db->Bpgsiz-bp->upper);
		p->lower-=sizeof(int16);
		memcpy(p->index,bp->index,p->lower-BPHEADSIZ);
		if((char*)bp!=buf) free(bp);
		putpage(db->pcc,(char*)p,F_DIRTY);
		return true;
	}else{
	    char buf[BLOCKSIZ];
		struct bpg* bp=(struct bpg*)buf;
		int i,k=0;
		idx=(struct bidx*)((char*)p+p->index[ipos-1]);
		if(idx->fap&B_PKOFLOW){
		    delopg(db,*((uint64*)((char*)idx+BIHEADSIZ)));
		}
		if(db->Bpgsiz>BLOCKSIZ) bp=malloc(db->Bpgsiz);
		bp->upper=db->Bpgsiz;
		for(i=0;i<inum;i++){
		    if(i!=ipos-1){
			    idx=(struct bidx*)((char*)p+p->index[i]);
				isiz=idxsiz(db,idx);
				bp->upper-=isiz;
				memcpy((char*)bp+bp->upper,idx,isiz);
				bp->index[k++]=bp->upper;
			}
		}
		p->upper=bp->upper;
		p->lower-=sizeof(int16);
		memcpy((char*)p+p->upper,(char*)bp+bp->upper,db->Bpgsiz-bp->upper);
		memcpy(p->index,bp->index,p->lower-BPHEADSIZ);
		if((char*)bp!=buf) free(bp);
		putpage(db->pcc,(char*)p,F_DIRTY);
		return true;
	}
}
int delopg(struct bdb *db,uint64 off){
	char buf[OPHEADSIZ];
	struct opg *op=(struct opg*)buf;
	seekread(db->fd,buf,OPHEADSIZ,off);
	DEBUG(printf("%s:off=%d ref=%d\n",__func__,off,OPREFERRED(op)));
	if(!OPREFERRED(op))
	    bdeloverflowpage(db,off,0,0x7fffffff);//a trick..
	else{
	    OPUNREFER(op);
	    seekwrite(db->fd,buf,OPHEADSIZ,off);
	}
	 return true;
}
/**
    return -1 indicate find fail
	       >=0 indicate the position of the value
*/
int findspecial(struct bdb *db,uint64 off,char *val,int vsiz,struct bc *cursor){
    struct vdpg *vp;
	char *ptr,*nval;
	int i,vnum,inum=0;
	char buf[BLOCKSIZ];
	nval=buf;
	if(vsiz>BLOCKSIZ) nval=malloc(vsiz); 
	while(off!=0){
	    DEBUG(printf("%s:search vdpg=%d\n",__func__,off));
	    vp=(struct vdpg*)getpage(db->lcc,off,F_READ);
	    vnum=(vp->lower-VDHEADSIZ)/sizeof(int16);
	    for(i=0;i<vnum;i++){
	        ptr=vp->index[i]+(char*)vp;
			inum++;
		    if(vsiz==*((int*)ptr)){
		        ptr+=sizeof(int32);
		        if(vsiz>VP_OSIZ){
			        bgetoverflowpage(db,*((uint64*)ptr),0,nval,vsiz);
				}else{
				    nval=ptr;
				}
				if(strncmp(val,nval,vsiz)==0){
				    BC(cursor,off,i,vp);
					if(nval!=buf) free(nval);
					return inum;
				}
			}
		}
		off=vp->nextpg;
		putpage(db->lcc,(char*)vp,0);
	}
	if(nval!=buf) free(nval);
	return -1;
}
/**
   return: -1 indicates the special value does not exsit,
            0 indicates all values were deleted,
			others indicate 1 or more than 1 values remained
*/
int delspecialval(struct bdb *db,uint64 off,char *val,int vsiz,uint64 *noff){
    struct bc cursor;
	int ret=-1,vpos;
	*noff=off;
	if((vpos=findspecial(db,off,val,vsiz,&cursor))!=-1){
	    struct vdpg *vp=cursor.p;
		int ipos=cursor.idxoff;
		int vnum=(vp->lower-VDHEADSIZ)/sizeof(int16);
		char *ptr=(char*)vp+vp->index[ipos];
		ret=0;
		DEBUG(printf("%s:off=%d cursor.offset=%d vnum=%d ipos=%d\n",__func__,off,cursor.offset,vnum,ipos));
		if(vsiz>VP_OSIZ){
		    bdeloverflowpage(db,*((uint64*)(ptr+sizeof(int32))),0,vsiz);
		}
		if(vnum==1){
		    if(((struct vdpg*)cursor.p)->prepg!=0){
		        vp=(struct vdpg*)getpage(db->lcc,((struct vdpg*)(cursor.p))->prepg,F_WRITE);
			    vp->nextpg=((struct vdpg*)cursor.p)->nextpg;
				putpage(db->lcc,(char*)vp,F_DIRTY);
				ret=1;
			}
			if(((struct vdpg*)cursor.p)->nextpg!=0){
			    vp=(struct vdpg*)getpage(db->lcc,((struct vdpg*)cursor.p)->nextpg,F_WRITE);
				vp->prepg=((struct vdpg*)cursor.p)->prepg;
				putpage(db->lcc,(char*)vp,F_DIRTY);
				if(cursor.offset==off) *noff=((struct vdpg*)cursor.p)->nextpg;
				ret=1;
			}
			pthread_mutex_lock(&db->iter.lock);
			if((db->iter.flag&BI_INUSE)&&db->iter.voffset==cursor.offset){
			    if(((struct vdpg*)cursor.p)->nextpg!=0){
			        db->iter.voffset=((struct vdpg*)cursor.p)->nextpg;
					db->iter.voff=0;
				}
				else if(vpos>0){
				    db->iter.idxoff+=(db->iter.flag&BI_BACKWORD)?-1:1;
					db->iter.voffset=0;
				}
			}
			pthread_mutex_unlock(&db->iter.lock);
			delpage(db->lcc,(char*)cursor.p);
			bfreepg(db,cursor.offset);
		}else{
			char buf[BLOCKSIZ];
			struct vdpg *bvp=(struct vdpg*)buf;
			int i,siz,k=0;
			if(db->Bpgsiz>BLOCKSIZ) bvp=malloc(db->Bpgsiz);
			bvp->upper=db->Bpgsiz;
			for(i=0;i<vnum;i++){
			    if(i!=cursor.idxoff){
				    ptr=(char*)vp+vp->index[i];
					siz=*((int32*)ptr)+sizeof(int32);
					if(*((int32*)ptr)>VP_OSIZ){
					    siz=sizeof(uint64)+sizeof(int32);
					}
					bvp->upper-=siz;
					memcpy((char*)bvp+bvp->upper,ptr,siz);
					bvp->index[k++]=bvp->upper;
				}
			}
			vp->upper=bvp->upper;
			vp->lower-=sizeof(int16);
			memcpy((char*)vp+vp->upper,(char*)bvp+bvp->upper,db->Bpgsiz-bvp->upper);
			memcpy(vp->index,bvp->index,vp->lower-VDHEADSIZ);
			if((char*)bvp!=buf) free(bvp);
			pthread_mutex_lock(&db->iter.lock);
			if((db->iter.flag&BI_INUSE)&&db->iter.voffset==cursor.offset&&db->iter.voff==vnum-1){
				if(vp->nextpg!=0){
				    db->iter.voffset=vp->nextpg;
					db->iter.voff=0;
				}else{
				    db->iter.idxoff+=db->iter.flag&BI_BACKWORD?-1:1;
					db->iter.voffset=0;
				}
			}
			pthread_mutex_unlock(&db->iter.lock);
			putpage(db->lcc,(char*)vp,F_DIRTY);
			ret=1;
		}
	}
	DEBUG(printf("%s:noff=%d ret=%d\n",__func__,*noff,ret));
	return ret;
}
int deldupval(struct bdb *db,uint64 off){
    struct vdpg *vp;
	int vnum,i,vsiz;
	char *ptr;
	while(off!=0){
	    vp=(struct vdpg*)getpage(db->lcc,off,F_WRITE);
		vnum=(vp->lower-VDHEADSIZ)/sizeof(int16);
		for(i=0;i<vnum;i++){
		    ptr=(char*)vp+vp->index[i];
			vsiz=*((int32*)ptr);
			if(vsiz>VP_OSIZ){
			    ptr+=sizeof(int32);
				bdeloverflowpage(db,*((uint64*)ptr),0,vsiz);
			}
		}
		bfreepg(db,off);
		off=vp->nextpg;
		delpage(db->lcc,(char*)vp);
	}
	return true;
}
int biterinit(struct bdb* db,char *key,int ksiz,int flag){
    struct biter *it=&db->iter;
	if(db->Broot==0){
	    it->flag&=~BI_INUSE;
		return false;
	}
	it->flag=BI_INUSE|(flag&BI_INCLUDE);
	if(key!=NULL&&ksiz!=0){
	    it->flag|=(flag&BI_BACKWORD);
	    jump(db,key,ksiz,&it->offset,&it->idxoff,flag);
		DEBUG(printf("%s:offset=%d idxoff=%d\n",__func__,it->offset,it->idxoff));
	}else if(flag&BI_BACKWORD){
	    it->flag|=BI_BACKWORD;
	    it->offset=db->Brmost;
		it->idxoff=INT16_MAX;
	}else{
	    it->offset=db->Blmost;
		it->idxoff=0;
	}
	it->voff=0;
	it->voffset=0;
	DEBUG(printf("%s:it->offset=%d idxoff=%d flag=%d\n",__func__,it->offset,it->idxoff,it->flag));
	return true;
}
int jump(struct bdb *db,char *key,int ksiz,uint64 *offset,int *idxoff,int flag){
    *offset=db->Broot;
	struct bpg *p;
	int ipos;
	while(*offset!=0&&!ISBLEAF(*offset)){
	    p=(struct bpg*)getpage(db->pcc,*offset,F_READ);
		ipos=pgsearch(db,p,key,ksiz);
		DEBUG(printf("%s:offset=%d\n",__func__,*offset));
		if(ipos<1) *offset=p->left;
		else *offset=((struct bidx*)((char*)p+p->index[ipos-1]))->offset;
		putpage(db->pcc,(char*)p,0);
	}
	struct blf *lf=(struct blf*)getpage(db->lcc,OFFSET(*offset),F_READ);
	int hit=0;
	*idxoff=lfsearch(db,lf,key,ksiz,&hit);
	DEBUG(printf("%s:offset=%d idxoff=%d\n",__func__,*offset,*idxoff));
	if(hit&&!(flag&BI_INCLUDE)){
	    if(!(flag&BI_BACKWORD)){
		    int rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
			if(*idxoff>=rnum-1){
			    *offset=lf->next;
				*idxoff=0;
			}else{
			    (*idxoff)++;
			}
		}else{
		    if(*idxoff==0){
			    *offset=lf->pre;
				*idxoff=INT16_MAX;
			}else{
			    (*idxoff)--;
			}
		}
	}
	putpage(db->lcc,(char*)lf,0);
	return hit;
}
int getkey(struct bdb *db,struct brec *rec,char *key,int ksiz){
    int flag=rec->fap&B_FMASK&~B_VDUP;
	int siz=ksiz>rec->ksiz?rec->ksiz:ksiz;
	if(flag==0){
		memcpy(key,(char*)rec+BRHEADSIZ,siz);
	}else if(flag==B_VOFLOW){
	    memcpy(key,(char*)rec+BRHEADSIZ+sizeof(uint64),siz);
	}else if(flag==B_PKOFLOW){
	    int pksiz=db->osiz-BRHEADSIZ-sizeof(uint64);
		int cpsiz=siz>pksiz?pksiz:siz;
		memcpy(key,(char*)rec+BRHEADSIZ+sizeof(uint64),cpsiz);
		if(cpsiz<siz){
		    bgetoverflowpage(db,*((uint64*)((char*)rec+BRHEADSIZ)),0,key+cpsiz,siz-cpsiz);
		}
	}else{
	    printf("%s:error:fap=%x\n",__func__,rec->fap);
		exit(1);
	}
	return siz;
}
int biternext(struct bdb *db,struct bstring **key,struct bstring **val){
    struct biter *it=&db->iter;
	pthread_mutex_lock(&it->lock);
	if(!(it->flag&BI_INUSE)||it->offset==0){
	    printf("%s:no more rec!\n",__func__);
		pthread_mutex_unlock(&it->lock);
		return false;
	}
	struct blf *lf=(struct blf*)getpage(db->lcc,OFFSET(it->offset),F_READ);
	int rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
	DEBUG(printf("%s:rnum=%d\n",__func__,rnum));
	if(!(it->flag&BI_BACKWORD)){//default forword
	    if(it->idxoff>=rnum){//next leaf
		    if(lf->next==0){
			    putpage(db->lcc,(char*)lf,0);//no next leaf
				it->flag&=~BI_INUSE;
				DEBUG(printf("%s:it->idxoff=%d rnum=%d\n",__func__,it->idxoff,rnum));
				pthread_mutex_unlock(&it->lock);
				return false;
			}
			it->offset=lf->next;
			it->idxoff=0;
			it->voffset=0;
			putpage(db->lcc,(char*)lf,0);
			lf=(struct blf*)getpage(db->lcc,OFFSET(it->offset),F_READ);
			rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
		}
		DEBUG(printf("%s:idxoff=%d rnum=%d\n",__func__,it->idxoff,rnum));
		struct brec *rec=(struct brec*)((char*)lf+lf->index[it->idxoff]);
		int flag=rec->fap&B_FMASK;
		if(flag&B_VDUP){
			if(it->voffset==0){
				it->voffset=rec->sop;
				it->voff=0;
			}
			if(*key==NULL)	*key=strnew(rec->ksiz);
			else if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
		    getkey(db,rec,(*key)->array,rec->ksiz);
			(*key)->siz=rec->ksiz;
			struct vdpg *vp=(struct vdpg*)getpage(db->lcc,it->voffset,F_READ);
			int vnum=(vp->lower-VDHEADSIZ)/sizeof(int16);
			if(it->voff>=vnum){
			    printf("%s:error:voff=%d vnum=%d\n",__func__,it->voff,vnum);
				exit(1);
			}
			char *ptr=(char*)vp+vp->index[it->voff];
			if(*val==NULL)	*val=strnew(*((int32*)ptr));
			else if((*val)->siz<*((int32*)ptr)) stradjust(*val,*((int32*)ptr));
			(*val)->siz=*((int32*)ptr);
			ptr+=sizeof(int32);
			if((*val)->siz>VP_OSIZ){
				bgetoverflowpage(db,*((uint64*)ptr),0,(*val)->array,(*val)->siz);
			}else{
				memcpy((*val)->array,ptr,(*val)->siz);
			}
			it->voff++;
			if(it->voff>=vnum){//move to the next value page...
				if(vp->nextpg!=0) it->voffset=vp->nextpg;
				else{
					it->voffset=0;
					it->voff=0;
					it->idxoff++;
				}
			}
			putpage(db->lcc,(char*)vp,0);
			putpage(db->lcc,(char*)lf,0);
		}else{
			if(flag==0){
				if(*val==NULL){
					*val=strcreate((char*)rec+BRHEADSIZ+rec->ksiz,rec->sop);
				}else{
					if((*val)->siz<rec->sop)
						stradjust(*val,rec->sop);
					memcpy((*val)->array,(char*)rec+BRHEADSIZ+rec->ksiz,rec->sop);
					(*val)->siz=rec->sop;
				}
				if(*key==NULL)
					*key=strcreate((char*)rec+BRHEADSIZ,rec->ksiz);
				else{
					if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
					memcpy((*key)->array,(char*)rec+BRHEADSIZ,rec->ksiz);
					(*key)->siz=rec->ksiz;
				}
				//memcpy(val->array,(char*)rec+BRHEADSIZ+rec->ksiz,rec->sop);
				//memcpy(key->array,(char*)rec+BRHEADSIZ,rec->ksiz);
			}else if(flag==B_VOFLOW){
				if(*key==NULL)
					*key=strcreate((char*)rec+BRHEADSIZ+sizeof(uint64),rec->ksiz);
				else{
					if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
					memcpy((*key)->array,(char*)rec+BRHEADSIZ+sizeof(uint64),rec->ksiz);
					(*key)->siz=rec->ksiz;
				}
			    //memcpy(key->array,(char*)rec+BRHEADSIZ+sizeof(uint64),rec->ksiz);
				if(*val==NULL)	*val=strnew(rec->sop);
				else if((*val)->siz<rec->sop) stradjust(*val,rec->sop);
				(*val)->siz=rec->sop;
				bgetoverflowpage(db,*((uint64*)((char*)rec+BRHEADSIZ)),0,(*val)->array,(*val)->siz);
			}else if(flag==B_PKOFLOW){
				int pksiz=db->osiz-BRHEADSIZ-sizeof(uint64);
				char buf[BLOCKSIZ];
				char *kvb=buf;
				if(*key==NULL)	*key=strnew(rec->ksiz);
				else if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
				(*key)->siz=rec->ksiz;
				if(BLOCKSIZ<rec->ksiz-pksiz+rec->sop) kvb=malloc(rec->ksiz-pksiz+rec->sop);
				memcpy((*key)->array,(char*)rec+BRHEADSIZ+sizeof(uint64),pksiz);
			 	bgetoverflowpage(db,*((uint64*)((char*)rec+BRHEADSIZ)),0,kvb,rec->ksiz-pksiz+rec->sop);
				memcpy((*key)->array+pksiz,kvb,rec->ksiz-pksiz);
				if(*val==NULL)	*val=strcreate(kvb+rec->ksiz-pksiz,rec->sop);
				else{
					if((*val)->siz<rec->sop) stradjust(*val,rec->sop);
					memcpy((*val)->array,kvb+rec->ksiz-pksiz,rec->sop);
					(*val)->siz=rec->sop;
				}
				if(kvb!=buf) free(kvb);
			}else{
			    printf("%s:error:fap=%x\n",__func__,rec->fap);
				exit(1);
			}
			putpage(db->lcc,(char*)lf,0);
			it->idxoff++;
			it->voffset=0;
			it->voff=0;
		}
	}else{
	    if(it->idxoff<0){
		    if(lf->pre==0){
			    putpage(db->lcc,(char*)lf,0);
				it->flag&=~BI_INUSE;
				pthread_mutex_unlock(&it->lock);
				return false;
			}
			it->offset=lf->pre;
			putpage(db->lcc,(char*)lf,0);
			lf=(struct blf*)getpage(db->lcc,OFFSET(it->offset),F_READ);
			rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
			it->idxoff=INT16_MAX;
		}
		//rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
		DEBUG(printf("%s:idxoff=%d rnum=%d offset=%d\n",__func__,it->idxoff,rnum,it->offset));
		if(it->idxoff==INT16_MAX){
		    it->idxoff=rnum-1;
			it->voffset=0;
			it->voff=0;
		}
		struct brec *rec=(struct brec*)((char*)lf+lf->index[it->idxoff]);
		int flag=rec->fap&B_FMASK;
		if(flag&B_VDUP){
		    if(it->voffset==0){
			    it->voffset=rec->sop;
				it->voff=0;
			}
			if(*key==NULL)	*key=strnew(rec->ksiz);
			else if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
			(*key)->siz=rec->ksiz;
		    getkey(db,rec,(*key)->array,rec->ksiz);
			struct vdpg* vp=(struct vdpg*)getpage(db->lcc,it->voffset,F_READ);
			int vnum=(vp->lower-VDHEADSIZ)/sizeof(int16);
			if(it->voff>=vnum){
			    printf("%s:error..:voff=%d vnum=%d\n",__func__,it->voff,vnum);
				exit(1);
			}
			char *ptr=vp->index[it->voff]+(char*)vp;
			//stradjust(val,*((int32*)ptr));
			if(*val==NULL)	*val=strnew(*((int32*)ptr));
			else if((*val)->siz<*((int32*)ptr)) stradjust(*val,*((int32*)ptr));
			(*val)->siz=*((int32*)ptr);
			ptr+=sizeof(int32);
			if((*val)->siz>VP_OSIZ){
			    bgetoverflowpage(db,*((uint64*)ptr),0,(*val)->array,(*val)->siz);
			}else{
			    memcpy((*val)->array,ptr,(*val)->siz);
			}
			it->voff++;
			if(it->voff==vnum){
			    if(vp->nextpg==0){
				    it->idxoff--;
					it->voffset=0;
					it->voff=0;
				}else{
				    it->voffset=vp->nextpg;
					it->voff=0;
				}
			}
			putpage(db->lcc,(char*)vp,0);
			putpage(db->lcc,(char*)lf,0);
		}else{
		    //stradjust(key,rec->ksiz);
		    //stradjust(val,rec->sop);
		    if(flag==0){
				if(*key==NULL)	*key=strcreate((char*)rec+BRHEADSIZ,rec->ksiz);
				else{
					if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
					memcpy((*key)->array,(char*)rec+BRHEADSIZ,rec->ksiz);
					(*key)->siz=rec->ksiz;
				}
				if(*val==NULL)	*val=strcreate((char*)rec+BRHEADSIZ+rec->ksiz,rec->sop);
				else{
					if((*val)->siz<rec->sop) stradjust(*val,rec->sop);
					memcpy((*val)->array,(char*)rec+BRHEADSIZ+rec->ksiz,rec->sop);
					(*val)->siz=rec->sop;
				}
			    //memcpy(key->array,(char*)rec+BRHEADSIZ,rec->ksiz);
			    //memcpy(val->array,(char*)rec+BRHEADSIZ+rec->ksiz,rec->sop);
			}else if(flag==B_VOFLOW){
			    //memcpy(key->array,(char*)rec+BRHEADSIZ+sizeof(uint64),rec->ksiz);
				if(*key==NULL)
					*key=strcreate((char*)rec+BRHEADSIZ+sizeof(uint64),rec->ksiz);
				else{
					if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
					memcpy((*key)->array,(char*)rec+BRHEADSIZ+sizeof(uint64),rec->ksiz);
					(*key)->siz=rec->ksiz;
				}
				if(*val==NULL)	*val=strnew(rec->sop);
				else if((*val)->siz<rec->sop) stradjust(*val,rec->sop);
				(*val)->siz=rec->sop;
			    bgetoverflowpage(db,*((uint64*)((char*)rec+BRHEADSIZ)),0,(*val)->array,(*val)->siz);
			}else if(flag==B_PKOFLOW){
			    int pksiz=db->osiz-BRHEADSIZ-sizeof(uint64);
				char buf[BLOCKSIZ];
				char *kvb=buf;
				if(*key==NULL)	*key=strnew(rec->ksiz);
				else if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
				(*key)->siz=rec->ksiz;
				if(BLOCKSIZ<rec->sop+rec->ksiz-pksiz) kvb=malloc(rec->sop+rec->ksiz-pksiz);
				memcpy((*key)->array,(char*)rec+BRHEADSIZ+sizeof(uint64),pksiz);
				bgetoverflowpage(db,*((uint64*)((char*)rec+BRHEADSIZ)),0,kvb,rec->sop+rec->ksiz-pksiz);
				memcpy((*key)->array+pksiz,kvb,rec->ksiz-pksiz);
				if(*val==NULL)	*val=strcreate(kvb+rec->ksiz-pksiz,rec->sop);
				else{
					if((*val)->siz<rec->sop) stradjust(*val,rec->sop);
					memcpy((*val)->array,kvb+rec->ksiz-pksiz,rec->sop);
					(*val)->siz=rec->sop;
				}
				if(kvb!=buf) free(kvb);
			}else{printf("%s:error:fap=%x\n",__func__,rec->fap);exit(1);}
			putpage(db->lcc,(char*)lf,0);
			it->idxoff--;
			it->voffset=0;
			it->voff=0;
		}
	}
	pthread_mutex_unlock(&it->lock);
	return true;
}
int biterclose(struct bdb *db){
    pthread_mutex_lock(&db->iter.lock);
    db->iter.flag&=~BI_INUSE;
	pthread_mutex_unlock(&db->iter.lock);
	return true;
}
int btraverse(struct bdb* db){
    uint64 stack[1000];
	char buf[BLOCKSIZ];
	char *key=buf;
	int ksiz=BLOCKSIZ;
	int beg=0,end=1;
	if(db->Broot==0){
	   printf("%s:root is null\n",__func__);
	   exit(1);
	}
	struct bpg* p;
	struct blf *lf;
	uint64 off=db->Broot;
	stack[0]=off;
	while(beg<end){
	    off=stack[beg];
		if(off==0){
		    printf("%s:beg=%d end=%d\n",__func__,beg,end);
			exit(1);
		}
		if(ISBLEAF(off)){
		    lf=(struct blf*)getpage(db->lcc,OFFSET(off),F_READ);
			int rnum=(lf->lower-BLHEADSIZ)/sizeof(int16);
			int i;
			DEBUG(printf("%s:leaf=%d rnum=%d\n",__func__,off,rnum));
			for(i=0;i<rnum;i++){
			    struct brec *rec=(struct brec*)((char*)lf+lf->index[i]);
				int flag=rec->fap&B_FMASK&~B_VDUP;
				if(ksiz<rec->ksiz){
				    if(key==buf) key=malloc(rec->ksiz);
					else key=realloc(key,rec->ksiz);
					ksiz=rec->ksiz;
				}
				if(flag==B_PKOFLOW){
				    int pksiz=db->osiz-BRHEADSIZ-sizeof(uint64);
				    memcpy(key,(char*)rec+BRHEADSIZ+sizeof(uint64),pksiz);
					bgetoverflowpage(db,*((uint64*)((char*)rec+BRHEADSIZ)),0,key+pksiz,rec->ksiz-pksiz);
				}else if(flag==B_VOFLOW){
				    memcpy(key,(char*)rec+BRHEADSIZ+sizeof(uint64),rec->ksiz);
				}else if(flag==0) memcpy(key,(char*)rec+BRHEADSIZ,rec->ksiz);
				else{
				    printf("%s:fap=%x\n",__func__,rec->fap);
					exit(1);
				}
				DEBUG(int k=xatoi(key,rec->ksiz);
				printf("%s:%d rec key=%d %s\n",__func__,i,k,rec->fap&B_VDUP?"vd":""));
			}
			putpage(db->lcc,(char*)lf,0);
		}else{
		    p=(struct bpg*)getpage(db->pcc,off,F_READ);
			int inum=(p->lower-BPHEADSIZ)/sizeof(int16);
			int i;
			DEBUG(printf("%s:page=%d inum=%d left=%d\n",__func__,off,inum,p->left));
			stack[end++]=p->left;
			for(i=0;i<inum;i++){
			    struct bidx *idx=(struct bidx*)((char*)p+p->index[i]);
				int flag=idx->fap&B_FMASK;
				if(ksiz<idx->ksiz){
				    if(key==buf) key=malloc(idx->ksiz);
					else key=realloc(key,idx->ksiz);
					ksiz=idx->ksiz;
				}
				if(flag==0){
				    memcpy(key,(char*)idx+BIHEADSIZ,idx->ksiz);
				}else if(flag==B_PKOFLOW){
				    int pksiz=db->osiz-BIHEADSIZ-sizeof(uint64);
					memcpy(key,(char*)idx+BIHEADSIZ+sizeof(uint64),pksiz);
					bgetoverflowpage(db,*((uint64*)((char*)idx+BIHEADSIZ)),0,key+pksiz,idx->ksiz-pksiz);
				}else{
				    printf("%s:idx->fap=%x\n",__func__,idx->fap);
					exit(1);
				}
				DEBUG(int k=xatoi(key,idx->ksiz);
				printf("%s:%d idx key=%d offset=%d\n",__func__,i,k,idx->offset));
				stack[end++]=idx->offset;
				if(end>sizeof(stack)/sizeof(uint64)){
				   printf("%s:overflow!!!!!!\n",__func__);
				   exit(1);
				}
			}
		}
		beg++;
	}
	return true;
}
int bshrinkpage(struct bdb *db){
	if(db->Bnfpg>0){
	    char buf[BLOCKSIZ];
	    uint64 *parray=(uint64*)buf;
		int i=0;
		uint64 pgno;
		if(db->Bnfpg*sizeof(uint64)>BLOCKSIZ)
		    parray=malloc(db->Bnfpg*sizeof(uint64));
		if(db->Bfstart>=0){
		    if(db->Bfend<0){
			    memcpy(parray,db->fpa,db->Bfnum*sizeof(uint64));
				i=db->Bfnum;
			}
			else if(db->Bfend>db->Bfstart){
			    i=db->Bfend-db->Bfstart;
			    memcpy(parray,db->fpa+db->Bfstart,i*sizeof(uint64));
			}
		    else if(db->Bfend<db->Bfstart){
			    i=db->Bfnum-db->Bfstart;
			    memcpy(parray,db->fpa+db->Bfstart,i*sizeof(uint64));
			    memcpy(parray+i,db->fpa,db->Bfend*sizeof(uint64));
			    i+=db->Bfend;
		    }
		}
		if(db->Bfread!=0){
		    struct freepage fp;
			pgno=db->Bfread;
			//printf("%s:fread=%d\n",__func__,db->Bfread);
			while(pgno!=0&&i<db->Bnfpg){
			    seekread(db->fd,(char*)&fp,FREEHEAD,pgno);
				if(pgno<db->Bmaxpg)
				    parray[i++]=pgno;
				pgno=fp.nextpg;
				
			}
		}
		//printf("%s:i=%d db->Bnfpg=%d maxpg=%d\n",__func__,i,db->Bnfpg,db->Bmaxpg);
		if(db->Bflast!=0){
		    struct freepage fp;
			pgno=db->Bflast;
			//printf("%s:flast pgno=%d\n",__func__,pgno);
			while(pgno!=0&&i<db->Bnfpg){
			    seekread(db->fd,(char*)&fp,sizeof(fp),pgno);
				if(pgno<db->Bmaxpg)
				    parray[i++]=pgno;
				//printf("%s:link pgno=%d i=%d\n",__func__,pgno,i);
				pgno=fp.nextpg;
			}
		}
		//printf("%s:new pgno=%d i=%d db->Bnfpg=%d\n",__func__,pgno,i,db->Bnfpg);
		sort64(parray,i);
		/*printf("%s:i=%d db->Bnfpg=%d\n",__func__,i,db->Bnfpg);
		int x;
		for(x=0;x<i;x++){
		    printf("%s:%d pgno=%d\n",__func__,x,parray[x]);
		}*/
		while(parray[i-1]>=db->Bmaxpg) i--;
		//printf("%s:i=%d\n",__func__,i);
		while(i>0&&parray[i-1]==db->Bmaxpg-db->Bpgsiz){
		    i--;
			db->Bmaxpg-=db->Bpgsiz;
		}
		DEBUG(printf("%s:i=%d\n",__func__,i));
		if(i==0){
		    db->Bflast=db->Bfread=0;
			db->Bfstart=-1;
			db->Bnfpg=0;
			return true;
		}
		
		if(db->Bnfpg>i){
		    i--;
		    db->Bmaxpg=parray[i];
		}
		int cp=i<db->Bfnum?i:db->Bfnum;
		memcpy(db->fpa,parray,cp*sizeof(uint64));
		db->Bfstart=0;
		db->Bfend=(cp==db->Bfnum)?-1:cp;
		while(cp<i-1){
		    struct freepage fp={FREEPAGEMAGIC};
			fp.nextpg=parray[cp+1];
			//printf("%s:cp=%d pgno=%d\n",__func__,cp,parray[cp]);
			seekwrite(db->fd,(char*)&fp,sizeof(struct freepage),parray[cp]);
		    cp++;
		}
		if(cp==i-1){
		   struct freepage fp={FREEPAGEMAGIC};
		   fp.nextpg=0;
		   //printf("%s:write last pgno=%d\n",__func__,parray[cp]);
		   seekwrite(db->fd,(char*)&fp,sizeof(struct freepage),parray[cp]);
		   db->Bflast=0;
		   db->Bfread=parray[cp];
		}else{
		   db->Bflast=0;
		   db->Bfread=0;
		}
		db->Bnfpg=i;
		DEBUG(printf("%s:nfpg=%d fstart=%d fend=%d flast=%d fread=%d maxpg=%d\n",__func__,db->Bnfpg,db->Bfstart,db->Bfend,db->Bflast,db->Bfread,db->Bmaxpg));
	}
	return true;
}
int bdelete(struct bdb* db){
    int dlen=strlen(db->dname);
    char *buf=malloc(dlen+1);
	if(buf==NULL) return false;
	memcpy(buf,db->dname,dlen);
	buf[dlen]='\0';
	bclose(db);
	unlink(buf);
	return true;
}
int bdbtrans_beg(struct bdb* db){
    if(db->iter.flag&BI_INUSE) return false;
    pthread_rwlock_wrlock(&db->pcc->lock);
	pthread_rwlock_wrlock(&db->lcc->lock);
	cachesyn(db->pcc);
	cachesyn(db->lcc);
	TRANSACTION_SET(db->pcc);
	TRANSACTION_SET(db->lcc);
	if(db->newpg==NULL) STACK_NEW(db->newpg,uint64);
	if(db->delpg==NULL) STACK_NEW(db->delpg,uint64);
	BDB_TRANSACTION_SET(db);
	return true;
}
int bdbtrans_commit(struct bdb *db){
    uint64 pgno;
	BDB_TRANSACTION_UNSET(db);
	STACK_POP(db->delpg,uint64,pgno);
	while(pgno!=0){
	    bfreepg(db,pgno);
		STACK_POP(db->delpg,uint64,pgno);
	}
	cachesyn(db->pcc);
	cachesyn(db->lcc);
	TRANSACTION_UNSET(db->pcc);
	TRANSACTION_UNSET(db->lcc);
	STACK_CLEAR(db->newpg);
	STACK_CLEAR(db->delpg);
	pthread_rwlock_unlock(&db->pcc->lock);
	pthread_rwlock_unlock(&db->lcc->lock);
	return true;
}
int bdbtrans_abort(struct bdb *db){
    uint64 pgno;
	BDB_TRANSACTION_UNSET(db);
	STACK_POP(db->newpg,uint64,pgno);
	while(pgno!=0){
	    bfreepg(db,pgno);
		if(pgno==OFFSET(db->Broot)) db->Broot=0;
		if(pgno==OFFSET(db->Blmost)){
		    struct blf *l=(struct blf*)getpage(db->lcc,F_READ,pgno);
			db->Blmost=l->next;
			putpage(db->lcc,(char*)l,0);
		}
		if(pgno==OFFSET(db->Brmost)){
		    struct blf *l=(struct blf*)getpage(db->lcc,F_READ,pgno);
			db->Brmost=l->pre;
			putpage(db->lcc,(char*)l,0);
		}
		STACK_POP(db->newpg,uint64,pgno);
	}
	cacheabort(db->lcc);
	cacheabort(db->pcc);
	STACK_CLEAR(db->newpg);
	STACK_CLEAR(db->delpg);
	TRANSACTION_UNSET(db->pcc);
	TRANSACTION_UNSET(db->lcc);
	pthread_rwlock_unlock(&db->pcc->lock);
	pthread_rwlock_unlock(&db->lcc->lock);
	return true;
}
int bdbtrans_cancel(struct bdb* db){
    TRANSACTION_UNSET(db->pcc);
	TRANSACTION_UNSET(db->lcc);
	BDB_TRANSACTION_UNSET(db);
    pthread_rwlock_unlock(&db->pcc->lock);
	pthread_rwlock_unlock(&db->lcc->lock);
	return true;
}
